Durable Objects Setup

// features/applications/applications.ts
import { sqliteTable, text } from "drizzle-orm/sqlite-core";
import { defineTable } from "@quickback/compiler";

export const applications = sqliteTable("applications", {
  id: text("id").primaryKey(),
  candidateId: text("candidate_id").notNull(),
  jobId: text("job_id").notNull(),
  stage: text("stage").notNull(),
  organizationId: text("organization_id").notNull(),
  // ── quickback:audit (compiler-managed — edits are validated, not merged) ──
  createdAt: text("created_at").notNull().default('1970-01-01T00:00:00.000Z').$defaultFn(() => new Date().toISOString()),
  modifiedAt: text("modified_at").notNull().default('1970-01-01T00:00:00.000Z').$defaultFn(() => new Date().toISOString()).$onUpdate(() => new Date().toISOString()),
  createdBy: text("created_by"),
  modifiedBy: text("modified_by"),
  deletedAt: text("deleted_at"),
  deletedBy: text("deleted_by"),
});

export default defineTable(applications, {
  firewall: [{ field: 'organizationId', equals: 'ctx.activeOrgId' }],
  realtime: {
    enabled: true,           // Enable realtime broadcasts for this table
    onInsert: true,          // Broadcast on INSERT (default: true)
    onUpdate: true,          // Broadcast on UPDATE (default: true)
    onDelete: true,          // Broadcast on DELETE (default: true)
    access: { roles: ["recruiter", "hiring-manager"] }, // Audience — MANDATORY, fail-closed
    fields: ["id", "candidateId", "stage"],   // Fields to include (optional)
    scopeKeyFrom: "events:{eventId}",         // Sugar for scopeTable + scopeField (optional)
    emitFromActions: true,                    // Custom-action writes auto-emit after commit (optional)
  },
  // ... guards, crud
});

providers: {
  database: defineDatabase("cloudflare-d1", {
    realtime: {
      wsTicket: {
        access: { authzRole: "attendee" },   // authz relationship role (legacy shorthand: role: "attendee")
        // access: { roles: ["member+"] },   // org-membership roles — verifies the org role AND that the
        //                                   // scope row is in the caller's ACTIVE org (org column required)
        // access: { fga: { relation: "viewer", object: "event:{id}" } }, // FGA relation on the scope row —
        //                                   // same evaluator + object template as read-path access:{ fga }
        requestField: "eventId",
        scopeTable: "events",
      },
    },
  }),
}

realtime: {
  wsTicket: {
    namespace: "eventScope",   // authorize via this namespace's full multi-lane `via`
    requestField: "eventId",
    scopeTable: "events",
  },
},

export const execute: ActionExecutor = async ({ db, ctx, input, realtime }) => {
  // After creating a record
  await realtime.insert('applications', newApplication, ctx.activeOrgId!);

  // After updating a record
  await realtime.update('applications', newApplication, oldApplication, ctx.activeOrgId!);

  // After deleting a record
  await realtime.delete('applications', { id: applicationId }, ctx.activeOrgId!);

  return { success: true };
};

await realtime.broadcast("event-page-updated", payload, {
  scopeKey: `events:${input.eventId}`,
});

await realtime.broadcast("dm", payload, {
  userId: recipientId,
});

await realtime.insert('applications', {
  id: 'app_123',
  candidateId: 'cnd_456',
  stage: 'applied'
}, ctx.activeOrgId!);

{
  "type": "postgres_changes",
  "table": "applications",
  "eventType": "INSERT",
  "schema": "public",
  "new": {
    "id": "app_123",
    "candidateId": "cnd_456",
    "stage": "applied"
  },
  "old": null
}

await realtime.update('applications',
  { id: 'app_123', stage: 'interview' },  // new
  { id: 'app_123', stage: 'screening' },  // old
  ctx.activeOrgId!
);

{
  "type": "postgres_changes",
  "table": "applications",
  "eventType": "UPDATE",
  "schema": "public",
  "new": { "id": "app_123", "stage": "interview" },
  "old": { "id": "app_123", "stage": "screening" }
}

await realtime.delete('applications',
  { id: 'app_123' },
  ctx.activeOrgId!
);

{
  "type": "postgres_changes",
  "table": "applications",
  "eventType": "DELETE",
  "schema": "public",
  "new": null,
  "old": { "id": "app_123" }
}

await realtime.broadcast('screening-complete', {
  applicationId: 'app_123',
  candidateId: 'cnd_456',
  stage: 'interview'
}, ctx.activeOrgId!, {
  targetRoles: ['AUTHENTICATED'],
});

{
  "type": "broadcast",
  "event": "screening-complete",
  "payload": {
    "applicationId": "app_123",
    "candidateId": "cnd_456",
    "stage": "interview"
  }
}

// Only this user receives the notification
await realtime.insert('notifications', newNotification, ctx.activeOrgId!, {
  userId: ctx.userId,
  targetRoles: ['AUTHENTICATED'],
});

// Notify a hiring manager of a new application
await realtime.broadcast('application-received', {
  applicationId: 'app_123',
  jobId: 'job_789'
}, ctx.activeOrgId!, {
  userId: hiringManagerId,
  targetRoles: ['AUTHENTICATED'],
});

// Only hiring managers and recruiters receive this broadcast
await realtime.insert('applications', application, ctx.activeOrgId!, {
  targetRoles: ['hiring-manager', 'recruiter']
});

// Interviewers only see applications assigned to them
await realtime.update('applications', newRecord, oldRecord, ctx.activeOrgId!, {
  targetRoles: ['hiring-manager']
});

// ONE send reaches this guest whether they're connected via their Better Auth
// session or a sessionless scope link — no per-identity double-send.
await realtime.broadcast('chat.frame', payload, { scopeKey }, {
  targetTags: [`scope:guest:${guestId}`],
  targetRoles: ['AUTHENTICATED'],
});

// Reach a person under either identity explicitly (deduped per socket):
await realtime.broadcast('rsvp.updated', payload, { scopeKey }, {
  targetTags: [`scope:guest:${guestId}`, `user:${linkedUserId}`],
  targetRoles: ['AUTHENTICATED'],
});

providers: {
  database: {
    config: {
      realtime: {
        enabled: true,
        lease: '15m',   // default; also '900s', '1h', or a number of seconds
        // lease: false // disable (discouraged — reopens the revocation gap)
      },
    },
  },
}

// On a timer comfortably inside the lease (e.g. every 10 min for a 15m lease):
const { wsTicket } = await mintTicket();          // POST /broadcast/v1/ws-ticket
ws.send(JSON.stringify({ type: 'reauth', ticket: wsTicket }));

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === 'reauth_ok') {
    scheduleNextRenewal(msg.authExpiresAt);        // lease extended, roles refreshed
  }
  if (msg.type === 'error' && msg.code === 'REAUTH_FAILED') {
    // Ticket invalid/expired — mint a fresh one and retry before the lease lapses.
  }
};

ws.onclose = (event) => {
  if (event.code === 4401) {
    // Lease expired or identity drift: mint a fresh ticket and reconnect.
    reconnectWithFreshTicket();
  }
};

await realtime.insert('candidates', newCandidate, ctx.activeOrgId!, {
  maskingConfig: {
    phone: { type: 'phone', show: { roles: ['owner', 'hiring-manager', 'recruiter'] } },
    email: { type: 'email', show: { roles: ['owner', 'hiring-manager', 'recruiter'] } },
    resumeUrl: { type: 'redact', show: { roles: ['owner', 'hiring-manager', 'recruiter'] } },
  },
});

maskingConfig: {
  phone: {
    type: 'phone',
    show: { roles: ['hiring-manager'], or: 'owner' }  // Hiring manager OR owner sees unmasked
  },
}

// 1. Get a ws-ticket from your API (requires session auth)
const response = await fetch('/broadcast/v1/ws-ticket', {
  method: 'POST',
  headers: {
    Authorization: `Bearer ${sessionToken}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({ eventId: 'evt_123' }),
});
const { wsTicket } = await response.json();

// 2. Connect with ticket as URL param
const ws = new WebSocket(
  `wss://api.yourdomain.com/broadcast/v1/websocket?ws_ticket=${wsTicket}`
);

ws.onopen = () => {
  console.log('Connected and authenticated!');
  // No auth message needed — pre-authenticated at upgrade
};

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);

  // Handle CRUD events
  if (msg.type === 'postgres_changes') {
    const { table, eventType, new: newRecord, old: oldRecord } = msg;

    if (eventType === 'INSERT') {
      addRecord(table, newRecord);
    } else if (eventType === 'UPDATE') {
      updateRecord(table, newRecord);
    } else if (eventType === 'DELETE') {
      removeRecord(table, oldRecord.id);
    }
  }

  // Handle custom broadcasts
  if (msg.type === 'broadcast') {
    const { event, payload } = msg;

    if (event === 'screening-complete') {
      refreshApplication(payload.applicationId);
    } else if (event === 'interview-scheduled') {
      showInterviewNotification(payload.applicationId);
    }
  }
};

[[durable_objects.bindings]]
name = "BROADCASTER"
class_name = "Broadcaster"

[[migrations]]
tag = "v1"
new_classes = ["Broadcaster"]

                              1. POST /ws-ticket
┌───────────────────────────────────────────────┐    ┌─────────────────┐
│              Quickback Worker                 │ ◄──│  Browser Client  │
│                                               │ ──►│                  │
│  ┌──────────┐  DO binding  ┌────────────────┐ │    └───────┬─────────┘
│  │ Hono API │ ───────────► │ Broadcaster DO │ │            │
│  └──────────┘  (in-process)└───────┬────────┘ │  2. WS ?ws_ticket=...
│                                    │          │            │
└────────────────────────────────────┼──────────┘            │
                                     │◄──────────────────────┘
                              WebSocket (verified at upgrade)

// Good - broadcast after successful insert
const [newApp] = await db.insert(applications).values(data).returning();
await realtime.insert('applications', newApp, ctx.activeOrgId!);

// Bad - don't broadcast before confirming success
await realtime.insert('applications', data, ctx.activeOrgId!);
await db.insert(applications).values(data);  // Could fail!

// Good - minimal payload
await realtime.update('applications',
  { id: record.id, stage: 'interview' },
  { id: record.id, stage: 'screening' },
  ctx.activeOrgId!
);

// Avoid - sending entire record with large content
await realtime.update('applications', fullRecord, oldRecord, ctx.activeOrgId!);

// Hiring pipeline stage completed
await realtime.broadcast('pipeline-stage-complete', {
  applicationId: application.id,
  stages: ['applied', 'screening', 'interview'],
  currentStage: 'offer',
  candidateId: application.candidateId
}, ctx.activeOrgId!);

// services/realtime/screening.ts
import { defineRealtime } from '@quickback/compiler';

export default defineRealtime({
  name: 'screening',
  events: ['started', 'progress', 'completed', 'failed'],
  description: 'Candidate screening pipeline events',
});

import { createRealtime } from '../lib/realtime';

export const execute: ActionExecutor = async ({ ctx, input }) => {
  const realtime = createRealtime(ctx.env);

  // Type-safe custom event methods
  await realtime.screening.started({
    applicationId: input.applicationId,
  }, ctx.activeOrgId!);

  // Progress updates
  await realtime.screening.progress({
    applicationId: input.applicationId,
    percent: 50,
    step: 'background-check',
  }, ctx.activeOrgId!);

  // Completion
  await realtime.screening.completed({
    applicationId: input.applicationId,
    result: 'passed',
    duration: 1234,
  }, ctx.activeOrgId!);

  return { success: true };
};

{
  "type": "broadcast",
  "event": "screening:started",
  "payload": {
    "applicationId": "app_123"
  }
}

// services/realtime/notifications.ts
export default defineRealtime({
  name: 'notifications',
  events: ['interview-reminder', 'offer-update', 'application-received'],
  description: 'Recruitment notification events',
});

// services/realtime/presence.ts
export default defineRealtime({
  name: 'presence',
  events: ['joined', 'left', 'reviewing', 'idle'],
  description: 'Who is reviewing which candidate',
});

const realtime = createRealtime(ctx.env);

// Notification events
await realtime.notifications['interview-reminder']({ ... }, ctx.activeOrgId!);

// Presence events — who's reviewing which candidate
await realtime.presence.reviewing({ userId: ctx.userId, candidateId: 'cnd_456' }, ctx.activeOrgId!);