-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: consume github app event queue
- Loading branch information
Showing
10 changed files
with
171 additions
and
94 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,35 +1,42 @@ | ||
| import { EventEmitter } from 'eventemitter3'; | ||
| import MultiMap from 'mnemonist/multi-map'; | ||
|
|
||
| import { TQueueMessage } from './types'; | ||
| import { githubWorker, githubWebhookWorker } from './worker/github'; | ||
|
|
||
| export type IConsumeWorker<T> = ( | ||
| message: Message<T>, | ||
| env: IRuntimeEnv, | ||
| ctx: ExecutionContext, | ||
| ) => void; | ||
| import { BaseWorker } from './worker'; | ||
| import { GitHubAppWorker } from './worker/github'; | ||
|
|
||
| export class QueueConsumer { | ||
| private emitter = new EventEmitter(); | ||
| addWorker<T, K extends TQueueMessage['type']>( | ||
| constructor(public env: IRuntimeEnv, public ctx: ExecutionContext) {} | ||
|
|
||
| private workerMap = new MultiMap< | ||
| TQueueMessage['type'], | ||
| BaseWorker<TQueueMessage> | ||
| >(); | ||
|
|
||
| addWorker<T extends TQueueMessage, K extends TQueueMessage['type']>( | ||
| type: K, | ||
| handler: IConsumeWorker<T>, | ||
| worker: BaseWorker<T>, | ||
| ) { | ||
| this.emitter.on(type, handler); | ||
| this.workerMap.set(type, worker); | ||
| } | ||
|
|
||
| consume( | ||
| message: Message<TQueueMessage>, | ||
| env: IRuntimeEnv, | ||
| ctx: ExecutionContext, | ||
| ) { | ||
| const { body } = message; | ||
| const { type } = body; | ||
| consume(message: Message<TQueueMessage>) { | ||
| for (const w of this.workerMap.values()) { | ||
| w.consume(message); | ||
| } | ||
| } | ||
|
|
||
| this.emitter.emit(type, message, env, ctx); | ||
| async runAndWait() { | ||
| const promises = [] as Promise<void>[]; | ||
| for (const w of this.workerMap.values()) { | ||
| promises.push(w.run()); | ||
| } | ||
| return await Promise.allSettled(promises); | ||
| } | ||
| } | ||
|
|
||
| export const consumer = new QueueConsumer(); | ||
| consumer.addWorker('github-app', githubWorker); | ||
| consumer.addWorker('github-webhook', githubWebhookWorker); | ||
| export const createConsumer = (env: IRuntimeEnv, ctx: ExecutionContext) => { | ||
| const consumer = new QueueConsumer(env, ctx); | ||
| consumer.addWorker('github-app', new GitHubAppWorker(env, ctx)); | ||
| // consumer.addWorker('github-webhook', githubWebhookWorker); | ||
| return consumer; | ||
| }; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,49 +1,77 @@ | ||
| import { Webhooks } from '@octokit/webhooks'; | ||
| import { EmitterWebhookEventName } from '@octokit/webhooks'; | ||
| import chain from 'lodash/chain'; | ||
| import MultiMap from 'mnemonist/multi-map'; | ||
|
|
||
| import { initApp } from '@/github/app'; | ||
| import { webhookHandler } from '@/github/handler'; | ||
| import { setupWebhooksTemplate } from '@/github/handler'; | ||
| import { MarkdownContent } from '@/github/types'; | ||
| import { sendToDing } from '@/github/utils'; | ||
| import { GitHubKVManager } from '@/kv/github'; | ||
|
|
||
| import { IGitHubEventQueueMessage } from '../types'; | ||
|
|
||
| export const githubWorker = async ( | ||
| message: Message<IGitHubEventQueueMessage>, | ||
| env: IRuntimeEnv, | ||
| ctx: ExecutionContext, | ||
| ) => { | ||
| const { body } = message; | ||
| const { botId, type, data } = body; | ||
| import { BaseWorker } from '.'; | ||
|
|
||
| const githubKVManager = new GitHubKVManager(); | ||
| const setting = await githubKVManager.getAppSettingById(botId); | ||
| export class GitHubAppWorker extends BaseWorker<IGitHubEventQueueMessage> { | ||
| async run() { | ||
| await Promise.allSettled( | ||
| chain(this.queue) | ||
| .groupBy((v) => v.body.botId) | ||
| .map(async (messages, botId) => { | ||
| const githubKVManager = new GitHubKVManager(); | ||
| const setting = await githubKVManager.getAppSettingById(botId); | ||
|
|
||
| if (setting && setting.githubSecret) { | ||
| const app = await initApp(setting); | ||
| if (setting && setting.githubSecret) { | ||
| const app = await initApp(setting); | ||
|
|
||
| await webhookHandler(botId, type, app.webhooks, ctx, data, true); | ||
| } else { | ||
| // todo logger | ||
| } | ||
| }; | ||
|
|
||
| export const githubWebhookWorker = async ( | ||
| message: Message<IGitHubEventQueueMessage>, | ||
| env: IRuntimeEnv, | ||
| ctx: ExecutionContext, | ||
| ) => { | ||
| const { body } = message; | ||
| const { botId, type, data } = body; | ||
|
|
||
| const githubKVManager = new GitHubKVManager(); | ||
| const setting = await githubKVManager.getSettingById(botId); | ||
|
|
||
| if (setting && setting.githubSecret) { | ||
| const webhooks = new Webhooks<{ octokit: undefined }>({ | ||
| secret: setting.githubSecret, | ||
| }); | ||
|
|
||
| await webhookHandler(botId, type, webhooks, ctx, data, true); | ||
| } else { | ||
| // todo logger | ||
| const results = new MultiMap<string, MarkdownContent>(); | ||
|
|
||
| setupWebhooksTemplate( | ||
| app.webhooks, | ||
| { setting }, | ||
| async ({ markdown, eventName }) => { | ||
| results.set(eventName, markdown); | ||
| }, | ||
| ); | ||
|
|
||
| await Promise.allSettled( | ||
| chain(messages) | ||
| .groupBy((v) => v.body.data.event) | ||
| .map(async (messages, eventName: EmitterWebhookEventName) => { | ||
| await Promise.all( | ||
| messages.map(async (message) => { | ||
| try { | ||
| const { data } = message.body; | ||
| await app.webhooks.receive({ | ||
| id: data.id, | ||
| name: data.event as any, | ||
| payload: data.payload, | ||
| }); | ||
| message.ack(); | ||
| } catch (error) { | ||
| console.error('github app worker error', error); | ||
| message.retry(); | ||
| } | ||
| }), | ||
| ); | ||
|
|
||
| const markdowns = results.get(eventName); | ||
| if (markdowns && markdowns.length > 0) { | ||
| // 只有特定内容的 content 要被合并起来 | ||
| await Promise.allSettled( | ||
| markdowns.map((markdown) => | ||
| sendToDing(markdown, eventName, setting), | ||
| ), | ||
| ); | ||
| } | ||
| }) | ||
| .value(), | ||
| ); | ||
| } else { | ||
| console.error('github app worker error: setting not found', botId); | ||
| } | ||
| }) | ||
| .value(), | ||
| ); | ||
| } | ||
| }; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| import { EventEmitter } from 'eventemitter3'; | ||
|
|
||
| export abstract class BaseWorker<T> { | ||
| protected emitter = new EventEmitter(); | ||
|
|
||
| queue: Message<T>[] = []; | ||
|
|
||
| constructor(public env: IRuntimeEnv, public ctx: ExecutionContext) {} | ||
|
|
||
| consume(message: Message<T>) { | ||
| this.queue.push(message); | ||
| } | ||
|
|
||
| abstract run(): Promise<void>; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters