template-app/modules/notes/core/subscribers.ts

// 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<P extends InProject> {
  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<string, unknown>
}

/** A subscriber that writes one audit event per record. */
export function auditing<P extends InProject>(shape: AuditShape<P>) {
  return async (event: { name: string; payload: P }, ctx: EventContext): Promise<void> => {
    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
// `<org id>/<resource id>` 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<Payloads["created"]>({
  action: NOTE_CREATED,
  actor: (p) => p.createdBy,
  target: note,
  occurrence: (p) => p.noteId,
  data: () => ({}),
})

const auditUpdated = auditing<Payloads["updated"]>({
  action: NOTE_UPDATED,
  actor: (p) => p.updatedBy,
  target: note,
  occurrence: (p) => `${p.noteId}:${p.version}`,
  data: (p) => ({ version: p.version }),
})

const auditDeleted = auditing<Payloads["deleted"]>({
  action: NOTE_DELETED,
  actor: (p) => p.deletedBy,
  target: note,
  occurrence: (p) => p.noteId,
  data: (p) => ({ attachmentCount: p.attachmentCount }),
})

const auditAdded = auditing<Payloads["added"]>({
  action: ATTACHMENT_ADDED,
  actor: (p) => p.attachedBy,
  target: note,
  occurrence: (p) => p.attachmentId,
  data: (p) => ({ attachmentId: p.attachmentId, fileId: p.fileId }),
})

const auditRemoved = auditing<Payloads["removed"]>({
  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<void> {
  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
    }
  })
}