Function queue

consumer/src/index.ts:23–54

(batch: MessageBatch<MessageBody>, env: Env, ctx: ExecutionContext)

export default {
	async queue(batch: MessageBatch<MessageBody>, env: Env, ctx: ExecutionContext) {
		const sentry = initSentryQueue(env, ctx)
		const db = await getDatabase(env)

		try {
			for (const message of batch.messages) {
				const actor = await actors.getActorById(db, new URL(message.body.actorId))
				if (actor === null) {
					console.warn(`actor ${message.body.actorId} is missing`)
					return
				}

				switch (message.body.type) {
					case MessageType.Inbox: {
						await handleInboxMessage(env, actor, message.body as InboxMessageBody)
						break
					}
					case MessageType.Deliver: {
						await handleDeliverMessage(env, actor, message.body as DeliverMessageBody)
						break
					}
					default:
						throw new Error('unsupported message type: ' + message.body.type)
				}
			}
		} catch (err: any) {
			if (sentry !== null) {
				sentry.captureException(err)
			}
			console.error(err.stack, err.cause)
		}
	},
}