events.js
2.29 KB
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
const EVENTS = 'vk_events'
const PAGE = 50
/**
* @typedef {{ id: number, type: string, task_id: string|null, created_at: Date }} AgentEvent
*/
/**
* 事件流的读取：发给我的事件 + 我所在公会的广播。`vk_events.id` 是游标。
* @param {{ db: import('knex').Knex }} deps
*/
export function createEvents({ db }) {
if (typeof db !== 'function') throw new Error('createEvents: db is required')
/**
* @param {import('./agents.js').Agent} agent
* @param {number} since
* @returns {Promise<AgentEvent[]>}
*/
async function after(agent, since) {
return db(EVENTS)
.where('id', '>', since)
.where((q) => q.where({ target_agent_id: agent.id }).orWhere((g) => g.whereNull('target_agent_id').where({ guild_id: agent.guild_id })))
.orderBy('id', 'asc')
.limit(PAGE)
.select('id', 'type', 'task_id', 'created_at')
}
/**
* `/home` 用：返回上次已读游标之后的事件，并把游标推进到本次返回的最后一条。
* @param {object} _ctx
* @param {{ agent: import('./agents.js').Agent }} input
* @returns {Promise<AgentEvent[]>}
*/
async function pullUnread(_ctx, { agent }) {
const row = await db('vk_agents').where({ id: agent.id }).first('events_cursor')
const events = await after(agent, row.events_cursor)
if (events.length > 0) await db('vk_agents').where({ id: agent.id }).where('events_cursor', '<', events[events.length - 1].id).update({ events_cursor: events[events.length - 1].id })
return events
}
/**
* `/events?since=` 用：补取，不推进游标。
* @param {object} _ctx
* @param {{ agent: import('./agents.js').Agent, since: number }} input
* @returns {Promise<AgentEvent[]>}
*/
async function listSince(_ctx, { agent, since }) {
return after(agent, since)
}
/**
* 在调用方的事务里写一条事件（任务状态机之外的来源，如打赏）。
* @param {object} _ctx
* @param {import('knex').Knex} trx
* @param {{ guild_id: string, task_id: string, type: string, target_agent_id: string|null }} input
*/
async function emit(_ctx, trx, { guild_id, task_id, type, target_agent_id }) {
await trx(EVENTS).insert({ guild_id, target_agent_id, type, task_id, payload: '{}', created_at: new Date() })
}
return { pullUnread, listSince, emit }
}