Under the Hood

// The datasets registered with the engine — one entry per live consumer.
// Adding a future consumer is a config change here plus a syncRef in
// convex/engine.ts, not new machinery.
//
// The engine serves a SINGLE live consumer: onlineStatus, the ≤2-min canary
// that keeps it exercised + proven for the v4.0 mapper. The three slow trackers
// moved to Neon stale-gated on-view reads.
export const SYNC_DATASETS = ['onlineStatus'] as const;
export type SyncDataset = (typeof SYNC_DATASETS)[number];

export const SYNC_DATASET_CONFIG: Record<
  SyncDataset,
  { cadenceFloorMs: number; tokenGroup: string }
> = {
  onlineStatus: { cadenceFloorMs: 60_000, tokenGroup: 'char-online' },
};

export const HEARTBEAT_MS = 20_000;
export const COLD_AFTER_MS = 60_000;
export const RETENTION_MS = 7 * 24 * 60 * 60_000;
export const STALE_RUNNING_MS = 3 * 60_000;

export function computeNextDueAt(
  minExpiresAt: number | null,
  cadenceFloorMs: number,
  now: number,
  random: () => number = Math.random,
): number {
  const due = Math.max(minExpiresAt ?? 0, now + cadenceFloorMs);
  return due + Math.floor(random() * SYNC_JITTER_MS);
}

export function minCacheWindow(windows: Array<number | null>): number | null {
  if (windows.length === 0 || windows.some((w) => w === null)) return null;
  return Math.min(...(windows as number[]));
}

// Convex is a regenerable projection of live ESI data keyed by the Neon
// identities (userId + characterId) — never the system of record, never a
// home for SDE/domain data.
//
// Since MIGRATE.B the engine serves a SINGLE live consumer — onlineStatus.
// The three slow trackers moved to Neon stale-gated on-view reads.

export default defineSchema({
  syncSubjects: defineTable({
    dataset: v.literal('onlineStatus'),
    userId: v.string(),
    status: v.union(v.literal('idle'), v.literal('running')),
    lastRequestedAt: v.number(),
    workId: v.union(v.string(), v.null()),
    nextDueAt: v.union(v.number(), v.null()),
    minExpiresAt: v.union(v.number(), v.null()),
    syncedCharacterIds: v.array(v.number()),
    lastFinishedAt: v.union(v.number(), v.null()),
    lastError: v.union(v.string(), v.null()),
    rlGroup: v.union(v.string(), v.null()),
    rlLimit: v.union(v.number(), v.null()),
    rlRemaining: v.union(v.number(), v.null()),
    rlUsed: v.union(v.number(), v.null()),
  })
    .index('by_user_dataset', ['userId', 'dataset'])
    .index('by_next_due', ['nextDueAt']),

  syncPresence: defineTable({
    dataset: v.literal('onlineStatus'),
    userId: v.string(),
    lastSeenAt: v.number(),
  })
    .index('by_user_dataset', ['userId', 'dataset'])
    .index('by_last_seen', ['lastSeenAt']),

  characterOnline: defineTable({
    userId: v.string(),
    characterId: v.number(),
    online: v.boolean(),
    etag: v.union(v.string(), v.null()),
  })
    .index('by_user', ['userId'])
    .index('by_user_character', ['userId', 'characterId']),
});

// The client half of the presence-gated sync engine: a visibility-gated heartbeat.
// While the tab is visible, beat every HEARTBEAT_MS; on hide, stop; on return,
// beat immediately so a stale view refreshes at once.

export function useSyncSubject(dataset: SyncDataset, characterIds: number[]) {
  const heartbeat = useMutation(api.engine.heartbeat);
  const characterIdsKey = characterIds.join(',');

  useEffect(() => {
    if (characterIdsKey === '') return;
    const characterIdsHint = characterIdsKey.split(',').map(Number);
    const beat = (reason: 'mount' | 'visible' | 'interval') =>
      void heartbeat({ dataset, characterIdsHint, reason });

    let timer: ReturnType<typeof setInterval> | null = null;
    const start = (reason: 'mount' | 'visible') => {
      beat(reason);
      timer = setInterval(() => beat('interval'), HEARTBEAT_MS);
    };
    const stop = () => {
      if (timer !== null) clearInterval(timer);
      timer = null;
    };

    if (document.visibilityState === 'visible') start('mount');
    document.addEventListener('visibilitychange', onVisibilityChange);
    return () => {
      stop();
      document.removeEventListener('visibilitychange', onVisibilityChange);
    };
  }, [dataset, characterIdsKey, heartbeat]);
}

// THE presence-gated sync engine — the one sanctioned presence/scheduling
// machinery. A subject (dataset × user) is refreshed on its dataset's cadence
// only while some visible tab is heartbeating it; cost scales with concurrently
// watched subjects, never with total linked characters.

const pool = new Workpool(components.workpool, { maxParallelism: 4 });

const rateLimiter = new RateLimiter(components.rateLimiter, {
  syncDispatch: { kind: 'token bucket', period: MINUTE, rate: 30, capacity: 10 },
});

const syncDatasetValidator = v.literal('onlineStatus');

const SYNC_REFS = {
  onlineStatus: internal.onlineStatusSync.syncUser,
} satisfies Record<SyncDataset, unknown>;

export const SCAN_DISPATCH_BATCH = 1024;

export const heartbeat = mutation({
  args: {
    dataset: syncDatasetValidator,
    characterIdsHint: v.array(v.number()),
    reason: v.union(v.literal('mount'), v.literal('visible'), v.literal('interval')),
  },
  handler: async (ctx, { dataset, characterIdsHint, reason }) => {
    const identity = await ctx.auth.getUserIdentity();
    if (identity === null) return;
    const userId = identity.subject;
    const now = Date.now();

    const presence = await getPresence(ctx.db, dataset, userId);
    if (presence === null) {
      await ctx.db.insert('syncPresence', { dataset, userId, lastSeenAt: now });
    } else {
      await ctx.db.patch(presence._id, { lastSeenAt: now });
    }

    if (reason === 'interval') return;

    let subject = await getSyncSubject(ctx.db, dataset, userId);
    if (subject === null) {
      const id = await ctx.db.insert('syncSubjects', {
        dataset,
        userId,
        status: 'idle',
        lastRequestedAt: 0,
        workId: null,
        nextDueAt: null,
        minExpiresAt: null,
        syncedCharacterIds: [],
        lastFinishedAt: null,
        lastError: null,
        rlGroup: null,
        rlLimit: null,
        rlRemaining: null,
        rlUsed: null,
      });
      subject = await ctx.db.get(id);
      if (subject === null) return;
    }

    if (!hasSyncTarget(subject.syncedCharacterIds, characterIdsHint)) return;
    if (isRunningFresh(subject.status, subject.lastRequestedAt, now)) return;
    if (!isStaleForImmediate(subject.minExpiresAt, subject.syncedCharacterIds, characterIdsHint, now)) {
      return;
    }
    await dispatch(ctx, subject, now);
  },
});

export const scan = internalMutation({
  args: {},
  handler: async (ctx) => {
    const now = Date.now();
    const due = await dueSubjects(ctx, now);
    for (const subject of due) {
      const presence = await getPresence(ctx.db, subject.dataset, subject.userId);
      if (isColdFromPresence(presence?.lastSeenAt ?? null, now)) {
        await ctx.db.patch(subject._id, { nextDueAt: null });
        continue;
      }
      if (isRunningFresh(subject.status, subject.lastRequestedAt, now)) continue;
      await dispatch(ctx, subject, now);
    }
    if (due.length === SCAN_DISPATCH_BATCH) {
      logBatchCapped('engine:scan', 'scan_batch_capped', due.length);
    }
  },
});

// Mounted once in the root layout: one subscription feeds every CharacterPortrait.
// The heartbeat hints only the active character; the sync action re-enumerates
// every linked character server-side.

export function OnlineStatusProvider({ children }: { children: ReactNode }) {
  if (convexClient === null) return <>{children}</>;
  return <OnlineStatusSubscribed>{children}</OnlineStatusSubscribed>;
}

function OnlineStatusSubscribed({ children }: { children: ReactNode }) {
  const view = useQuery(api.onlineStatus.forViewer);
  const map = useMemo(() => {
    const next = new Map<number, boolean>();
    for (const c of view?.characters ?? []) next.set(c.characterId, c.online);
    return next;
  }, [view]);

  return (
    <OnlineStatusContext.Provider value={map}>
      <Authenticated>
        <OnlineStatusHeartbeat />
      </Authenticated>
      {children}
    </OnlineStatusContext.Provider>
  );
}

function OnlineStatusHeartbeat() {
  const { session } = useAuth();
  useSyncSubject('onlineStatus', session ? [session.characterId] : []);
  return null;
}

// The one character portrait used everywhere — a round avatar with a live online
// dot. The dot is read from OnlineStatusProvider by characterId and lights only
// for the viewer's own characters.

export function CharacterPortrait({
  characterId,
  name,
  size,
  src,
  className,
  loading = 'lazy',
}: {
  characterId?: number;
  name: string;
  size: PortraitSize;
  src?: string;
  className?: string;
  loading?: 'lazy' | 'eager';
}) {
  const online = deriveOnlineState(useOnlineFlag(characterId ?? -1));
  const imageSrc = src ?? (characterId !== undefined ? characterPortraitUrl(characterId, 128) : '');

  return (
    <span className={cn('relative inline-block shrink-0', SIZE_CLASS[size], className)}>
      <img src={imageSrc} alt={name} width={size} height={size} className="size-full rounded-full border border-border-idle object-cover" />
      {online !== 'unknown' && (
        <StatusDot state={online} className="absolute top-[7%] right-[7%] translate-x-[2px] -translate-y-[2px]" />
      )}
    </span>
  );
}

// The viewer wire reads only characterOnline, so it re-fires only when that table
// changes — not on per-cycle dispatch/completion writes.

export const forViewer = query({
  args: {},
  handler: async (ctx) => {
    const identity = await ctx.auth.getUserIdentity();
    if (identity === null) return null;
    const userId = identity.subject;
    const docs = await ctx.db
      .query('characterOnline')
      .withIndex('by_user', (q) => q.eq('userId', userId))
      .collect();
    return {
      characters: docs.map((doc) => ({ characterId: doc.characterId, online: doc.online })),
    };
  },
});

export const applySyncResults = internalMutation({
  args: {
    userId: v.string(),
    generation: v.number(),
    enumeratedCharacterIds: v.array(v.number()),
    results: v.array(characterResultValidator),
    lastError: v.union(v.string(), v.null()),
    rlGroup: v.union(v.string(), v.null()),
    rlLimit: v.union(v.number(), v.null()),
    rlRemaining: v.union(v.number(), v.null()),
    rlUsed: v.union(v.number(), v.null()),
  },
  handler: async (ctx, args) => {
    const subject = await getSyncSubject(ctx.db, 'onlineStatus', args.userId);
    if (subject === null || subject.lastRequestedAt !== args.generation) return;

    const docs = await ctx.db.query('characterOnline').withIndex('by_user', (q) => q.eq('userId', args.userId)).collect();
    const byCharacter = new Map(docs.map((doc) => [doc.characterId, doc]));

    for (const result of args.results) {
      const window = await applyOnlineResult(ctx, args.userId, result, byCharacter.get(result.characterId));
      windowsByCharacter.set(result.characterId, window);
    }

    await stampSyncSubject(ctx, subject._id, [...windowsByCharacter.values()], args, now);
  },
});

async function applyOnlineResult(ctx: MutationCtx, userId: string, result: CharacterResult, existing: Doc<'characterOnline'> | undefined) {
  if (result.error !== null) return null;
  if (result.online === null) return result.expiresAt;
  if (existing === undefined) {
    await ctx.db.insert('characterOnline', { userId, characterId: result.characterId, online: result.online, etag: result.etag });
  } else if (existing.online !== result.online || existing.etag !== result.etag) {
    await ctx.db.patch(existing._id, { online: result.online, etag: result.etag });
  }
  return result.expiresAt;
}

// One run refreshes every linked character's online state for one user:
// heldState → Neon enumeration → eligibility + token vend + /online through the
// shared gate → one applySyncResults mutation.

export const syncUser = internalAction({
  args: { userId: v.string(), generation: v.number() },
  handler: async (ctx, { userId, generation }) => {
    const env = requireSyncEnv();
    const held = await ctx.runQuery(internal.onlineStatus.heldState, { userId });
    const heldByCharacter = new Map(held.map((h) => [h.characterId, h.etag]));
    const characters = await fetchEnumeratedCharacters(env, userId);

    const results: CharacterResult[] = [];
    const rl: RlSnapshot = { rlGroup: null, rlLimit: null, rlRemaining: null, rlUsed: null };

    for (const character of characters) {
      const heldEtag = heldByCharacter.get(character.characterId) ?? null;
      const outcome = await syncOnlineCharacter(env, character, heldEtag, rl);
      if (outcome.kind === 'skip') continue;
      results.push(outcome.result);
      if (outcome.kind === 'stop') break;
    }

    await ctx.runMutation(internal.onlineStatus.applySyncResults, {
      userId,
      generation,
      enumeratedCharacterIds: characters.map((c) => c.characterId),
      results,
      ...rl,
    });
  },
});

async function syncOnlineCharacter(env: SyncEnv, character: SyncCharacter, heldEtag: string | null, rl: RlSnapshot) {
  if (!canSyncOnline(character)) {
    return { kind: 'result', result: errorResult(character.characterId, 'reauth_required', heldEtag) };
  }

  const vend = await vendCharacterToken(env, character.characterId);
  if (vend.kind === 'skip') return { kind: 'skip' };
  if (vend.kind === 'reauth') return { kind: 'result', result: errorResult(character.characterId, 'reauth_required', heldEtag) };

  const read = await readEsiAuthed(`/characters/${character.characterId}/online`, vend.accessToken, heldEtag, rl);
  if (read.kind === 'unchanged') {
    return { kind: 'result', result: { characterId: character.characterId, online: null, etag: heldEtag, expiresAt, error: null } };
  }

  const online = parseOnlineBody(read.body);
  return { kind: 'result', result: { characterId: character.characterId, online, etag: read.etag, expiresAt, error: null } };
}

// Neon storage for the skill-queue tracker — replacing the live Convex skills
// datasets. The skills + skillqueue ESI endpoints cache 120s and queue completion
// is a pure timestamp flip derived client-side.

export const characterSkills = pgTable('character_skills', {
  characterId: bigint('character_id', { mode: 'number' }).primaryKey(),
  totalSp: bigint('total_sp', { mode: 'number' }).notNull(),
  unallocatedSp: bigint('unallocated_sp', { mode: 'number' }),
  queue: jsonb('queue').$type<SkillQueueEntry[]>().notNull().default([]),
});

export const characterSkillSyncs = pgTable('character_skill_syncs', {
  characterId: bigint('character_id', { mode: 'number' }).primaryKey(),
  lastRefreshedAt: timestamp('last_refreshed_at', { withTimezone: true }).notNull(),
  queueEtag: text('queue_etag'),
  skillsEtag: text('skills_etag'),
});

// Neon storage for the personal industry-jobs tracker — replacing the live
// Convex industry-jobs datasets. The ESI endpoint caches 300s, and a job's
// "ready" is derived client-side from end_date.

export const characterIndustryJobs = pgTable('character_industry_jobs', {
  characterId: bigint('character_id', { mode: 'number' }).primaryKey(),
  jobs: jsonb('jobs').$type<IndustryJob[]>().notNull().default([]),
});

export const characterIndustryJobSyncs = pgTable('character_industry_job_syncs', {
  characterId: bigint('character_id', { mode: 'number' }).primaryKey(),
  lastRefreshedAt: timestamp('last_refreshed_at', { withTimezone: true }).notNull(),
  jobsEtag: text('jobs_etag'),
});

// Corp jobs are keyed by (user_id, corporation_id), not corporation alone, because
// the board and role verdict are private to the signed-in user.
export const corpIndustryJobs = pgTable(
  'corp_industry_jobs',
  {
    userId: text('user_id').notNull(),
    corporationId: bigint('corporation_id', { mode: 'number' }).notNull(),
    jobs: jsonb('jobs').$type<IndustryJob[]>().notNull().default([]),
  },
  (t) => [primaryKey({ columns: [t.userId, t.corporationId] })],
);

export const corpIndustryJobSyncs = pgTable(
  'corp_industry_job_syncs',
  {
    userId: text('user_id').notNull(),
    corporationId: bigint('corporation_id', { mode: 'number' }).notNull(),
    lastRefreshedAt: timestamp('last_refreshed_at', { withTimezone: true }).notNull(),
    jobsEtag: text('jobs_etag'),
    syncError: text('sync_error'),
  },
  (t) => [primaryKey({ columns: [t.userId, t.corporationId] })],
);

// Skill-queue composition layer. It touches BOTH auth (token vend, scope reads)
// AND the skill-queue slice, so it lives above the slices.

export async function getSkillsForUserOnView(userId: string): Promise<ViewerSkillsResult> {
  const linked = await listLinkedCharacters(userId);
  const characterIds = linked.map((character) => character.characterId);
  const [dataMap, syncStates] = await Promise.all([
    getSkillsForCharacters(characterIds),
    Promise.all(characterIds.map((id) => readCharacterSyncState(id))),
  ]);
  after(() => refreshSkillsForUser(makeSkillsPort(), userId));

  const characters: ViewerSkills[] = characterIds.map((characterId, i) => ({
    characterId,
    data: dataMap.get(characterId) ?? null,
    lastRefreshedAt: syncStates[i]?.lastRefreshedAt?.getTime() ?? null,
  }));

  const nameMap = await getTypeNames([...skillIds]);
  return { characters, names };
}

// Personal industry-jobs composition layer. It reads cached boards immediately
// and fires a stale-gated write-behind refresh behind the response.

export async function getJobsForUserOnView(userId: string): Promise<ViewerJobsResult> {
  const linked = await listLinkedCharacters(userId);
  const characterIds = linked.map((character) => character.characterId);
  const [dataMap, syncStates] = await Promise.all([
    getJobsForCharacters(characterIds),
    Promise.all(characterIds.map((id) => readCharacterJobSyncState(id))),
  ]);
  after(() => refreshJobsForUser(makeJobsPort(), userId));

  const characters: ViewerJobs[] = characterIds.map((characterId, i) => ({
    characterId,
    data: dataMap.get(characterId) ?? null,
    lastRefreshedAt: syncStates[i]?.lastRefreshedAt?.getTime() ?? null,
  }));

  const nameMap = await getTypeNames([...new Set(jobTypeIds(characters))]);
  return { characters, names };
}

// Generic per-owner sync engine. The engine owns the mechanical dance every
// per-owner ESI→Neon slice clones: enumerate → stale-gate-before-vend → token /
// Director resolution → conditional fetch + plan → write-behind dispatch.

export interface OwnerSyncDescriptor<TOwner, TState, TSave> {
  now(): Date;
  enumerate(userId: string): Promise<EnumeratedOwner[]>;
  precondition?(owner: TOwner): Promise<boolean>;
  vendToken(characterId: number): Promise<string | null>;
  isStale(state: TState | null, now: Date): boolean;
  characterAxis?: OwnerAxis<TOwner>;
  corpAxis?: CorpOwnerAxis<TOwner>;
  readState(owner: TOwner): Promise<TState | null>;
  fetchAndPlan(owner: TOwner, accessToken: string, state: TState | null): Promise<PersistVerdict<TSave>>;
  save(owner: TOwner, payload: TSave): Promise<void>;
  stampFresh(owner: TOwner): Promise<void>;
  saveGateState?(owner: TOwner): Promise<void>;
}

export async function runOwnerSync<TOwner, TState, TSave>(
  descriptor: OwnerSyncDescriptor<TOwner, TState, TSave>,
  userId: string,
): Promise<void> {
  const owners = await descriptor.enumerate(userId);
  if (descriptor.characterAxis !== undefined) {
    await runCharacterPass(descriptor, descriptor.characterAxis, owners);
  }
  if (descriptor.corpAxis !== undefined) {
    await runCorpPass(descriptor, descriptor.corpAxis, userId, owners);
  }
}

// One owner, gated by staleness. resolveToken runs ONLY when the owner is stale.
async function syncOwner<TOwner, TState, TSave>(
  descriptor: OwnerSyncDescriptor<TOwner, TState, TSave>,
  owner: TOwner,
  resolveToken: () => Promise<TokenOutcome>,
): Promise<void> {
  if (descriptor.precondition !== undefined && !(await descriptor.precondition(owner))) return;

  const state = await descriptor.readState(owner);
  if (!descriptor.isStale(state, descriptor.now())) return;

  const token = await resolveToken();
  if (token.kind === 'skip') return;
  if (token.kind === 'needs_role') {
    await descriptor.saveGateState?.(owner);
    return;
  }

  const verdict = await descriptor.fetchAndPlan(owner, token.accessToken, state);
  switch (verdict.kind) {
    case 'skip':
      return;
    case 'stamp':
      await descriptor.stampFresh(owner);
      return;
    case 'needs_role':
      await descriptor.saveGateState?.(owner);
      return;
    case 'save':
      await descriptor.save(owner, verdict);
      return;
  }
}