talktime/backend/src/services/sync-worker.ts

import { randomUUID } from "node:crypto";
import { and, eq, lt, or, isNull, isNotNull } from "drizzle-orm";
import { db } from "../db/index.js";
import { syncJobs, members, chatRooms, teams } from "../db/schema.js";
import {
  createUser,
  createRoom,
  addUserToRoom,
  removeUserFromRoom,
  getUser,
  toMatrixUserId,
  toLocalpart,
  SynapseError,
} from "./synapse.js";
import { logAudit } from "../lib/audit.js";

const MAX_ATTEMPTS = 5;

type SyncJobPayload = {
  provision_matrix_user: { memberId: string; firstName: string; lastName: string };
  create_team_room: { teamId: string };
  join_team_room: { teamId: string; memberId: string };
  leave_team_room: { teamId: string; memberId: string };
};

type SyncJobType = keyof SyncJobPayload;

const handlers: {
  [K in SyncJobType]: (payload: SyncJobPayload[K]) => Promise<void>;
} = {
  provision_matrix_user: async ({ memberId, firstName, lastName }) => {
    const [member] = await db
      .select({ matrixUserId: members.matrixUserId })
      .from(members)
      .where(eq(members.id, memberId))
      .limit(1);
    if (!member) {
      throw new Error(`Member ${memberId} no longer exists`);
    }
    if (member.matrixUserId) {
      return; // already provisioned, idempotent no-op
    }

    const password = randomUUID().replace(/-/g, "");
    const displayName = `${firstName} ${lastName}`;

    for (let attempt = 0; attempt < 10; attempt++) {
      const localpart = toLocalpart();
      try {
        await createUser(localpart, { password, displayName });
        await db
          .update(members)
          .set({ matrixUserId: toMatrixUserId(localpart) })
          .where(eq(members.id, memberId));
        return;
      } catch (err) {
        if (err instanceof SynapseError && err.status === 400) {
          continue; // localpart taken, try next suffix
        }
        throw err;
      }
    }
    throw new Error(`Exhausted localpart suffixes for member ${memberId}`);
  },

  create_team_room: async ({ teamId }) => {
    const [existingRoom] = await db
      .select({ id: chatRooms.id })
      .from(chatRooms)
      .where(eq(chatRooms.teamId, teamId))
      .limit(1);
    if (existingRoom) {
      return; // already provisioned, idempotent no-op
    }

    const [team] = await db
      .select({ id: teams.id, organizationId: teams.organizationId, departmentId: teams.departmentId, name: teams.name })
      .from(teams)
      .where(eq(teams.id, teamId))
      .limit(1);
    if (!team) {
      throw new Error(`Team ${teamId} no longer exists`);
    }

    const { roomId } = await createRoom({ name: team.name });
    await db.insert(chatRooms).values({
      organizationId: team.organizationId,
      matrixRoomId: roomId,
      name: team.name,
      scope: "team",
      departmentId: team.departmentId,
      teamId: team.id,
    });
  },

  join_team_room: async ({ teamId, memberId }) => {
    const [room] = await db
      .select({ matrixRoomId: chatRooms.matrixRoomId })
      .from(chatRooms)
      .where(eq(chatRooms.teamId, teamId))
      .limit(1);
    if (!room) {
      throw new Error(`No chat room provisioned yet for team ${teamId}`);
    }

    const [member] = await db
      .select({ matrixUserId: members.matrixUserId })
      .from(members)
      .where(eq(members.id, memberId))
      .limit(1);
    if (!member?.matrixUserId) {
      throw new Error(`Member ${memberId} has no matrix account yet`);
    }

    await addUserToRoom(room.matrixRoomId, member.matrixUserId);
  },

  leave_team_room: async ({ teamId, memberId }) => {
    const [room] = await db
      .select({ matrixRoomId: chatRooms.matrixRoomId })
      .from(chatRooms)
      .where(eq(chatRooms.teamId, teamId))
      .limit(1);
    if (!room) {
      return; // no room ever provisioned, nothing to leave
    }

    const [member] = await db
      .select({ matrixUserId: members.matrixUserId })
      .from(members)
      .where(eq(members.id, memberId))
      .limit(1);
    if (!member?.matrixUserId) {
      return; // member was never provisioned, nothing to remove
    }

    await removeUserFromRoom(room.matrixRoomId, member.matrixUserId);
  },
};

export async function enqueueSyncJob<T extends SyncJobType>(
  organizationId: string,
  type: T,
  payload: SyncJobPayload[T],
): Promise<void> {
  await db.insert(syncJobs).values({
    organizationId,
    type,
    payload: JSON.stringify(payload),
  });
}

function backoffDelayMs(attempts: number): number {
  return Math.min(60_000 * 2 ** attempts, 30 * 60_000); // 1min, 2min, 4min... capped at 30min
}

export async function processPendingSyncJobs(): Promise<void> {
  const now = new Date();
  const pending = await db
    .select()
    .from(syncJobs)
    .where(
      and(
        eq(syncJobs.status, "pending"),
        or(isNull(syncJobs.updatedAt), lt(syncJobs.updatedAt, now)),
      ),
    )
    .limit(20);

  for (const job of pending) {
    const dueAt = new Date(job.updatedAt.getTime() + backoffDelayMs(job.attempts));
    if (job.attempts > 0 && dueAt > now) {
      continue; // still in backoff window
    }

    await db
      .update(syncJobs)
      .set({ status: "processing", updatedAt: new Date() })
      .where(eq(syncJobs.id, job.id));

    try {
      const handler = handlers[job.type as SyncJobType];
      if (!handler) {
        throw new Error(`Unknown sync job type: ${job.type}`);
      }
      await handler(JSON.parse(job.payload));
      await db
        .update(syncJobs)
        .set({ status: "done", updatedAt: new Date(), lastError: null })
        .where(eq(syncJobs.id, job.id));
    } catch (err) {
      const attempts = job.attempts + 1;
      const failed = attempts >= MAX_ATTEMPTS;
      await db
        .update(syncJobs)
        .set({
          status: failed ? "failed" : "pending",
          attempts,
          lastError: (err as Error).message,
          updatedAt: new Date(),
        })
        .where(eq(syncJobs.id, job.id));
    }
  }
}

export function startSyncWorker(intervalMs = 30_000): NodeJS.Timeout {
  return setInterval(() => {
    processPendingSyncJobs().catch((err) => {
      console.error("[sync-worker] failed to process jobs:", err);
    });
  }, intervalMs);
}

/**
 * Compares TalkTime's view of provisioned members against Synapse's actual
 * state and logs discrepancies (e.g. "member active here, deactivated/missing
 * on Synapse"). Read-only — corrective action is manual for now via the audit log.
 */
export async function runReconciliation(): Promise<void> {
  const provisioned = await db
    .select({
      id: members.id,
      organizationId: members.organizationId,
      status: members.status,
      matrixUserId: members.matrixUserId,
    })
    .from(members)
    .where(isNotNull(members.matrixUserId));

  for (const member of provisioned) {
    if (!member.matrixUserId) continue;

    let synapseUser;
    try {
      synapseUser = await getUser(member.matrixUserId);
    } catch (err) {
      console.warn(`[reconciliation] failed to check ${member.matrixUserId}:`, err);
      continue;
    }

    const expectedDeactivated = member.status !== "active";
    const synapseState = synapseUser === null
      ? "missing"
      : synapseUser.deactivated
        ? "deactivated"
        : "active";
    const mismatch = synapseUser === null || synapseUser.deactivated !== expectedDeactivated;

    if (mismatch) {
      await logAudit({
        organizationId: member.organizationId,
        actorUserId: null,
        action: "sync.discrepancy",
        targetType: "member",
        targetId: member.id,
        metadata: {
          matrixUserId: member.matrixUserId,
          localStatus: member.status,
          synapseState,
        },
      });
    }
  }
}

export function startReconciliationWorker(intervalMs = 10 * 60_000): NodeJS.Timeout {
  return setInterval(() => {
    runReconciliation().catch((err) => {
      console.error("[reconciliation] failed:", err);
    });
  }, intervalMs);
}