FilesExpand file tree

dev

/

event_manager.lua

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

397 lines (336 loc) · 15.8 KB

dev

/

event_manager.lua

Copy path

Top

File metadata and controls

• Code
• Blame

397 lines (336 loc) · 15.8 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

local state = require('opencode.state')

local config = require('opencode.config')

local ThrottlingEmitter = require('opencode.throttling_emitter')

local util = require('opencode.util')

--- @class EventInstallationUpdated

--- @field type "installation.updated"

--- @field properties {version: string}

--- @class EventLspClientDiagnostics

--- @field type "lsp.client.diagnostics"

--- @field properties {serverID: string, path: string}

--- @class EventMessageUpdated

--- @field type "message.updated"

--- @field properties {info: MessageInfo}

--- @class EventMessageRemoved

--- @field type "message.removed"

--- @field properties {sessionID: string, messageID: string}

--- @class EventMessagePartUpdated

--- @field type "message.part.updated"

--- @field properties {part: OpencodeMessagePart}

--- @class EventMessagePartRemoved

--- @field type "message.part.removed"

--- @field properties {sessionID: string, messageID: string, partID: string}

--- @class EventSessionCompacted

--- @field type "session.compacted"

--- @field properties {sessionID: string}

--- @class EventSessionIdle

--- @field type "session.idle"

--- @field properties {sessionID: string}

--- @class EventSessionUpdated

--- @field type "session.updated"

--- @field properties {info: Session}

--- @class EventSessionDeleted

--- @field type "session.deleted"

--- @field properties {info: Session}

--- @class EventSessionError

--- @field type "session.error"

--- @field properties {sessionID: string, error: table}

--- @class OpencodePermission

--- @field id string

--- @field type string

--- @field pattern string|string[]

--- @field sessionID string

--- @field messageID string

--- @field callID? string

--- @field title string

--- @field metadata table

--- @field time {created: number}

--- @class EventPermissionUpdated

--- @field type "permission.updated"

--- @field properties OpencodePermission

--- @class EventPermissionReplied

--- @field type "permission.replied"

--- @field properties {sessionID: string, permissionID: string, response: string}

--- @class EventFileEdited

--- @field type "file.edited"

--- @field properties {file: string}

--- @class EventFileWatcherUpdated

--- @field type "file.watcher.updated"

--- @field properties {file: string, event: "add"|"change"|"unlink"}

--- @class EventServerConnected

--- @field type "server.connected"

--- @field properties table

--- @class EventIdeInstalled

--- @field type "ide.installed"

--- @field properties {ide: string}

--- @class ServerStartingEvent

--- @field server_job table

--- @class ServerReadyEvent

--- @field server_job table

--- @field url string

--- @class ServerStoppedEvent

--- @class RestorePointCreatedEvent

--- @field restore_point RestorePoint

--- @alias OpencodeEventName

--- | "installation.updated"

--- | "lsp.client.diagnostics"

--- | "message.updated"

--- | "message.removed"

--- | "message.part.updated"

--- | "message.part.removed"

--- | "session.compacted"

--- | "session.idle"

--- | "session.updated"

--- | "session.deleted"

--- | "session.error"

--- | "permission.updated"

--- | "permission.replied"

--- | "file.edited"

--- | "file.watcher.updated"

--- | "server.connected"

--- | "ide.installed"

--- | "custom.server_starting"

--- | "custom.server_ready"

--- | "custom.server_stopped"

--- | "custom.restore_point.created"

--- | "custom.emit_events.started"

--- | "custom.emit_events.finished"

--- @class EventManager

--- @field events table<string, function[]> Event listener registry

--- @field server_subscription table|nil Subscription to server events

--- @field is_started boolean Whether the event manager is started

--- @field captured_events table[] List of captured events for debugging

--- @field throttling_emitter ThrottlingEmitter Throttle instance for batching events

local EventManager = {}

EventManager.__index = EventManager

--- Create a new EventManager instance

--- @return EventManager

function EventManager.new()

local self = setmetatable({

events = {},

server_subscription = nil,

is_started = false,

captured_events = {},

}, EventManager)

local throttle_ms = config.ui.output.rendering.event_throttle_ms

self.throttling_emitter = ThrottlingEmitter.new(function(events)

self:_on_drained_events(events)

end, throttle_ms)

return self

end

--- Subscribe to an event with type-safe callbacks using function overloads

--- @overload fun(self: EventManager, event_name: "installation.updated", callback: fun(data: EventInstallationUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "lsp.client.diagnostics", callback: fun(data: EventLspClientDiagnostics['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.updated", callback: fun(data: EventMessageUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.removed", callback: fun(data: EventMessageRemoved['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.part.updated", callback: fun(data: EventMessagePartUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.part.removed", callback: fun(data: EventMessagePartRemoved['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.compacted", callback: fun(data: EventSessionCompacted['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.idle", callback: fun(data: EventSessionIdle['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.updated", callback: fun(data: EventSessionUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.deleted", callback: fun(data: EventSessionDeleted['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.error", callback: fun(data: EventSessionError['properties']): nil)

--- @overload fun(self: EventManager, event_name: "permission.updated", callback: fun(data: EventPermissionUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "permission.replied", callback: fun(data: EventPermissionReplied['properties']): nil)

--- @overload fun(self: EventManager, event_name: "file.edited", callback: fun(data: EventFileEdited['properties']): nil)

--- @overload fun(self: EventManager, event_name: "file.watcher.updated", callback: fun(data: EventFileWatcherUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "server.connected", callback: fun(data: EventServerConnected['properties']): nil)

--- @overload fun(self: EventManager, event_name: "ide.installed", callback: fun(data: EventIdeInstalled['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.server_starting", callback: fun(data: ServerStartingEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.server_ready", callback: fun(data: ServerReadyEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.server_stopped", callback: fun(data: ServerStoppedEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.restore_point.created", callback: fun(data: RestorePointCreatedEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.emit_events.started", callback: fun(): nil)

--- @overload fun(self: EventManager, event_name: "custom.emit_events.finished", callback: fun(): nil)

--- @param event_name OpencodeEventName The event name to listen for

--- @param callback function Callback function to execute when event is triggered

function EventManager:subscribe(event_name, callback)

if not self.events[event_name] then

self.events[event_name] = {}

end

table.insert(self.events[event_name], callback)

end

--- Unsubscribe from an event with type-safe callbacks using function overloads

--- @overload fun(self: EventManager, event_name: "installation.updated", callback: fun(data: EventInstallationUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "lsp.client.diagnostics", callback: fun(data: EventLspClientDiagnostics['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.updated", callback: fun(data: EventMessageUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.removed", callback: fun(data: EventMessageRemoved['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.part.updated", callback: fun(data: EventMessagePartUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "message.part.removed", callback: fun(data: EventMessagePartRemoved['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.compacted", callback: fun(data: EventSessionCompacted['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.idle", callback: fun(data: EventSessionIdle['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.updated", callback: fun(data: EventSessionUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.deleted", callback: fun(data: EventSessionDeleted['properties']): nil)

--- @overload fun(self: EventManager, event_name: "session.error", callback: fun(data: EventSessionError['properties']): nil)

--- @overload fun(self: EventManager, event_name: "permission.updated", callback: fun(data: EventPermissionUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "permission.replied", callback: fun(data: EventPermissionReplied['properties']): nil)

--- @overload fun(self: EventManager, event_name: "file.edited", callback: fun(data: EventFileEdited['properties']): nil)

--- @overload fun(self: EventManager, event_name: "file.watcher.updated", callback: fun(data: EventFileWatcherUpdated['properties']): nil)

--- @overload fun(self: EventManager, event_name: "server.connected", callback: fun(data: EventServerConnected['properties']): nil)

--- @overload fun(self: EventManager, event_name: "ide.installed", callback: fun(data: EventIdeInstalled['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.server_starting", callback: fun(data: ServerStartingEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.server_ready", callback: fun(data: ServerReadyEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.server_stopped", callback: fun(data: ServerStoppedEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.restore_point.created", callback: fun(data: RestorePointCreatedEvent['properties']): nil)

--- @overload fun(self: EventManager, event_name: "custom.emit_events.started", callback: fun(): nil)

--- @overload fun(self: EventManager, event_name: "custom.emit_events.finished", callback: fun(): nil)

--- @param event_name OpencodeEventName The event name

--- @param callback function The callback function to remove

function EventManager:unsubscribe(event_name, callback)

local listeners = self.events[event_name]

if not listeners then

return

end

for i, cb in ipairs(listeners) do

if cb == callback then

table.remove(listeners, i)

break

end

end

end

---Callback from ThrottlingEmitter when the events are now ready to be processed.

---Collapses parts that are duplicated, making sure to replace earlier parts with later

---ones (but keeping the earlier position)

---@param events any

function EventManager:_on_drained_events(events)

self:emit('custom.emit_events.started', {})

if not config.ui.output.rendering.event_collapsing then

for _, event in ipairs(events) do

self:emit(event.type, event.properties)

end

self:emit('custom.emit_events.finished', {})

return

end

local collapsed_events = {}

local part_update_indices = {}

for i, event in ipairs(events) do

if event.type == 'message.part.updated' and event.properties.part then

local part_id = event.properties.part.id

if part_update_indices[part_id] then

-- vim.notify('collapsing: ' .. part_id .. ' text: ' .. vim.inspect(event.properties.part.text))

-- put this event in the earlier slot

-- move this newer part to the position of the original part

collapsed_events[part_update_indices[part_id]] = event

-- clear out this parts now unneeded position

collapsed_events[i] = nil

else

part_update_indices[part_id] = i

collapsed_events[i] = event

end

else

collapsed_events[i] = event

end

end

for i = 1, #events do

local event = collapsed_events[i]

if event then

self:emit(event.type, event.properties)

end

end

self:emit('custom.emit_events.finished', {})

end

--- Emit an event to all subscribers

--- @param event_name OpencodeEventName The event name

--- @param data any Data to pass to event listeners

function EventManager:emit(event_name, data)

local listeners = self.events[event_name]

if not listeners then

return

end

local event = { type = event_name, properties = data }

if config.debug.capture_streamed_events then

table.insert(self.captured_events, vim.deepcopy(event))

end

for _, callback in ipairs(listeners) do

local ok, result = util.pcall_trace(callback, data)

if not ok then

vim.notify('Error calling ' .. event_name .. ' listener: ' .. result, vim.log.levels.ERROR)

end

end

end

--- Start the event manager and begin listening to server events

function EventManager:start()

if self.is_started then

return

end

self.is_started = true

state.subscribe(

'opencode_server',

--- @param key string

--- @param current OpencodeServer|nil

--- @param prev OpencodeServer|nil

function(key, current, prev)

if current and current:get_spawn_promise() then

self:emit('custom.server_starting', { server_job = current })

current:get_spawn_promise():and_then(function(server)

self:emit('custom.server_ready', { server_job = server, url = server.url })

vim.defer_fn(function()

self:_subscribe_to_server_events(server)

end, 200)

end)

current:get_shutdown_promise():and_then(function()

self:emit('custom.server_stopped', {})

self:_cleanup_server_subscription()

end)

elseif prev and not current then

self:emit('custom.server_stopped', {})

self:_cleanup_server_subscription()

end

end

)

end

function EventManager:stop()

if not self.is_started then

return

end

self.is_started = false

self:_cleanup_server_subscription()

self.throttling_emitter:clear()

self.events = {}

end

--- Subscribe to server-sent events from the API

--- @param server table The server instance

function EventManager:_subscribe_to_server_events(server)

if not server.url then

return

end

self:_cleanup_server_subscription()

local api_client = state.api_client

local emitter = function(event)

self.throttling_emitter:enqueue(event)

end

self.server_subscription = api_client:subscribe_to_events(nil, emitter)

end

function EventManager:_cleanup_server_subscription()

if self.server_subscription then

pcall(function()

if self.server_subscription.shutdown then

self.server_subscription:shutdown()

elseif self.server_subscription.pid and type(self.server_subscription.pid) == 'number' then

vim.fn.jobstop(self.server_subscription.pid --[[@as integer]])

end

end)

self.server_subscription = nil

end

end

--- Get all event names that have subscribers

--- @return string[] List of event names

function EventManager:get_event_names()

local names = {}

for name, _ in pairs(self.events) do

table.insert(names, name)

end

return names

end

--- Get number of subscribers for an event

--- @param event_name OpencodeEventName The event name

--- @return number Number of subscribers

function EventManager:get_subscriber_count(event_name)

local listeners = self.events[event_name]

return listeners and #listeners or 0

end

function EventManager.setup()

state.event_manager = EventManager.new()

state.event_manager:start()

end

return EventManager