// What happens exactly once per change. // // The audit event and the usage fact hang off the outbox record the changing transaction wrote, and // the relay delivers a record exactly once per subscriber, so a note created by a retried request // is audited and metered once, and one whose process died before the relay ran still is, because // the record committed with the row. The platform's audit and usage APIs are HTTP and cannot join // the transaction: each call carries a key derived from the record, so a delivery the relay repeats // after a crash is the same event to the platform and not a second one. // // Ids, counts and versions only: no title and no body in an audit event (spec ยง9.4). import { type EventContext, internal } from "@conusai-kit/core" import { DATA_SOURCE } from "@conusai-kit/data" import { auditEventId, FILES_AUDIT, FILES_SCOPE, FILES_USAGE } from "@conusai-kit/files" import { ATTACHMENT_ADDED, ATTACHMENT_REMOVED, attachments, NOTE_CREATED, NOTE_DELETED, NOTE_UPDATED, NOTES_CREATED_METER, } from "../contract/index.js" interface InProject { orgId: string workspaceId: string projectId: string } /** * The organization a delivery belongs to comes from the container -- the scope the relay rebuilt * from the outbox record's own `orgId` -- and is compared with the payload's: a record that named * another tenant is refused here rather than written there. */ function assertSameOrg(ctx: EventContext, payload: InProject): void { const ambient = ctx.container.resolve(FILES_SCOPE) if (ambient.orgId !== payload.orgId) { throw internal(`a notes event of organization ${payload.orgId} was delivered in ${ambient.orgId}`) } } interface AuditShape

{ action: string actor: (payload: P) => string target: (payload: P) => { id: string; type: string } /** What makes two events of one action on one target two events: a version, a link's id. */ occurrence: (payload: P) => string data: (payload: P) => Record } /** A subscriber that writes one audit event per record. */ export function auditing

(shape: AuditShape

) { return async (event: { name: string; payload: P }, ctx: EventContext): Promise => { const { payload } = event assertSameOrg(ctx, payload) await ctx.container.resolve(FILES_AUDIT).log(shape.action, { actor: { kind: "human", id: shape.actor(payload) }, target: shape.target(payload), event_id: await auditEventId(shape.action, shape.occurrence(payload)), data: { ...shape.data(payload), projectId: payload.projectId }, }) } } type Payloads = { created: InProject & { noteId: string; createdBy: string } updated: InProject & { noteId: string; version: number; updatedBy: string } deleted: InProject & { noteId: string; deletedBy: string; attachmentCount: number } added: InProject & { noteId: string; attachmentId: string; fileId: string; attachedBy: string } removed: InProject & { noteId: string; attachmentId: string; fileId: string; removedBy: string } } // The organization has to be in the target or the in-app viewer could not show one // organization's events without another's (apps/web/src/server/audit.ts): a resource target is // `/` here, the id being what an outbox payload carries; the transport // writes the same rule with the slug it has, and the viewer matches either. const note = (payload: InProject & { noteId: string }) => ({ id: `${payload.orgId}/${payload.noteId}`, type: "note" }) const auditCreated = auditing({ action: NOTE_CREATED, actor: (p) => p.createdBy, target: note, occurrence: (p) => p.noteId, data: () => ({}), }) const auditUpdated = auditing({ action: NOTE_UPDATED, actor: (p) => p.updatedBy, target: note, occurrence: (p) => `${p.noteId}:${p.version}`, data: (p) => ({ version: p.version }), }) const auditDeleted = auditing({ action: NOTE_DELETED, actor: (p) => p.deletedBy, target: note, occurrence: (p) => p.noteId, data: (p) => ({ attachmentCount: p.attachmentCount }), }) const auditAdded = auditing({ action: ATTACHMENT_ADDED, actor: (p) => p.attachedBy, target: note, occurrence: (p) => p.attachmentId, data: (p) => ({ attachmentId: p.attachmentId, fileId: p.fileId }), }) const auditRemoved = auditing({ action: ATTACHMENT_REMOVED, actor: (p) => p.removedBy, target: note, occurrence: (p) => p.attachmentId, data: (p) => ({ attachmentId: p.attachmentId, fileId: p.fileId }), }) /** One audit event and one usage fact per note created. */ export async function onNoteCreated(event: { name: string; payload: Payloads["created"] }, ctx: EventContext) { await auditCreated(event, ctx) const usage = ctx.container.resolve(FILES_USAGE) await usage.track({ customer: { externalId: event.payload.orgId }, meter: NOTES_CREATED_METER, quantity: 1, idempotencyKey: `${NOTES_CREATED_METER}.${event.payload.noteId}`, }) // Awaited: a sink that sends per call is then durable before the relay marks the record published. await usage.flush() } export const onNoteUpdated = auditUpdated export const onNoteDeleted = auditDeleted export const onAttachmentAdded = auditAdded export const onAttachmentRemoved = auditRemoved /** * A file that stopped being stored takes its links with it, so a note never lists an attachment * nobody can open. The scope is the delivery's own organization and a system actor with the two * deeper ids of the record the files module wrote in that organization's transaction. */ export async function onFileDeleted( event: { name: string; payload: InProject & { fileId: string } }, ctx: EventContext, ): Promise { const { payload } = event assertSameOrg(ctx, payload) const ambient = ctx.container.resolve(FILES_SCOPE) const scope = { level: "project" as const, actor: ambient.actor, orgId: ambient.orgId, workspaceId: payload.workspaceId, projectId: payload.projectId, } await ctx.container.resolve(DATA_SOURCE).unitOfWork.run(scope, async (tx) => { for (;;) { const page = await tx.of(attachments).list({ where: [{ field: "fileId", op: "eq", value: payload.fileId }] }) for (const link of page.items) await tx.of(attachments).delete(link.id) if (!page.hasMore) break } }) }