/**
 * Phase 7 — retention-run state management.
 *
 * Three primitives the dispatcher / per-org worker call:
 *   - `claimRetentionRun(orgId)` → create an in-flight row, returns
 *     the new row's id; refuses if a younger-than-23h, uncompleted
 *     row already exists (prevents duplicate-fire when previous day's
 *     run is still draining a backlog).
 *   - `recordRetentionRunCheckpoint(runId, cursor, processed?)` → patch
 *     `lastCursor` + accumulate `processedCount` mid-run so a continuation
 *     can resume from the cursor.
 *   - `completeRetentionRun(runId, error?)` → patch `completedAt`.
 *
 * Plus a query helper used by the editor UI:
 *   - `getRetentionRunStatus(orgId)` → most recent run for the operator
 *     "longest-running cleanup" panel.
 */

import { ConvexError, v } from 'convex/values';

import { internalMutation, internalQuery, query } from '../_generated/server';
import { createAuditLog } from '../audit_logs/helpers';
import { getAuthUserIdentity } from '../lib/rls/auth/get_auth_user_identity';
import { isAdmin } from '../lib/rls/helpers/role_helpers';
import { getOrganizationMember } from '../lib/rls/organization/get_organization_member';

// Audit actions emitted by this file. Keep grep-able:
//   retention.run_started     — claimRetentionRun successful insert
//   retention.run_abandoned   — claimRetentionRun reaped a stale row
//   retention.run_completed   — completeRetentionRun without error
//   retention.run_failed      — completeRetentionRun with error

const STALE_RUN_AGE_MS = 23 * 60 * 60 * 1000; // 23h absolute ceiling
// A live worker checkpoints between every category and on continuation
// scheduling; a 30-min heartbeat gap means the worker crashed (or the host
// ran out of compute) and we shouldn't keep blocking the next dispatch
// for the remainder of the 23h absolute window.
const STALE_HEARTBEAT_MS = 30 * 60 * 1000;

const cursorValidator = v.object({
  category: v.string(),
  step: v.optional(v.string()),
  lastCreationTime: v.optional(v.number()),
});

export const claimRetentionRun = internalMutation({
  args: { organizationId: v.string() },
  returns: v.union(v.id('retentionRuns'), v.null()),
  handler: async (ctx, args) => {
    const now = Date.now();
    // Look for a recent in-flight run.
    const recent = await ctx.db
      .query('retentionRuns')
      .withIndex('by_organizationId_startedAt', (q) =>
        q.eq('organizationId', args.organizationId),
      )
      .order('desc')
      .first();

    if (recent && recent.completedAt === undefined) {
      const age = now - recent.startedAt;
      // Backwards-compat: rows persisted before lastHeartbeatAt landed
      // fall back to startedAt — they keep the 23h absolute behavior.
      const heartbeatAge =
        recent.lastHeartbeatAt !== undefined
          ? now - recent.lastHeartbeatAt
          : age;
      if (age < STALE_RUN_AGE_MS && heartbeatAge < STALE_HEARTBEAT_MS) {
        // In-flight + still heartbeating — skip this dispatch.
        return null;
      }
      // Either the absolute 23h deadline passed, or the worker stopped
      // checkpointing for >30 min (process crash / host OOM); mark the
      // row failed so the runs panel shows the gap, then claim a new
      // run.
      const abandonReason =
        age >= STALE_RUN_AGE_MS
          ? `Run abandoned (no completion within ${STALE_RUN_AGE_MS / 1000 / 60 / 60}h).`
          : `Run abandoned (no heartbeat within ${STALE_HEARTBEAT_MS / 1000 / 60} min).`;
      await ctx.db.patch(recent._id, {
        completedAt: now,
        lastError: recent.lastError ?? abandonReason,
      });
      // Audit the abandonment so the chain reflects the gap. Without
      // this, the only signal a run was reaped is the lastError column
      // on the runs panel — which is operator-readable but not part of
      // the tamper-evident chain.
      await createAuditLog(ctx, {
        organizationId: args.organizationId,
        actorId: 'system',
        actorType: 'system',
        action: 'retention.run_abandoned',
        category: 'data',
        resourceType: 'retentionRun',
        resourceId: String(recent._id),
        status: 'failure',
        errorMessage: recent.lastError ?? abandonReason,
        metadata: {
          startedAt: recent.startedAt,
          ageMs: age,
          heartbeatAgeMs: heartbeatAge,
        },
      });
    }

    const newRunId = await ctx.db.insert('retentionRuns', {
      organizationId: args.organizationId,
      startedAt: now,
      lastHeartbeatAt: now,
      processedCount: 0,
    });
    await createAuditLog(ctx, {
      organizationId: args.organizationId,
      actorId: 'system',
      actorType: 'system',
      action: 'retention.run_started',
      category: 'data',
      resourceType: 'retentionRun',
      resourceId: String(newRunId),
      status: 'success',
    });
    return newRunId;
  },
});

export const recordRetentionRunCheckpoint = internalMutation({
  args: {
    runId: v.id('retentionRuns'),
    cursor: v.optional(cursorValidator),
    processedDelta: v.optional(v.number()),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    const row = await ctx.db.get(args.runId);
    if (!row) return null;
    // Always touch the heartbeat — every checkpoint is a liveness signal
    // even when no cursor or processed delta is supplied.
    const patch: {
      lastCursor?: typeof row.lastCursor;
      processedCount?: number;
      lastHeartbeatAt: number;
    } = { lastHeartbeatAt: Date.now() };
    if (args.cursor !== undefined) {
      patch.lastCursor = args.cursor;
    }
    if (args.processedDelta !== undefined) {
      patch.processedCount = (row.processedCount ?? 0) + args.processedDelta;
    }
    await ctx.db.patch(args.runId, patch);
    return null;
  },
});

export const completeRetentionRun = internalMutation({
  args: {
    runId: v.id('retentionRuns'),
    error: v.optional(v.string()),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    const completedAt = Date.now();
    await ctx.db.patch(args.runId, {
      completedAt,
      lastError: args.error,
    });
    // Audit the run completion so the orchestration lifecycle is
    // captured in the tamper-evident chain alongside the per-deletion
    // rows. Look up orgId from the run row (single read; same row we
    // just patched).
    const row = await ctx.db.get(args.runId);
    if (row) {
      await createAuditLog(ctx, {
        organizationId: row.organizationId,
        actorId: 'system',
        actorType: 'system',
        action: args.error ? 'retention.run_failed' : 'retention.run_completed',
        category: 'data',
        resourceType: 'retentionRun',
        resourceId: String(args.runId),
        status: args.error ? 'failure' : 'success',
        errorMessage: args.error,
        metadata: {
          startedAt: row.startedAt,
          completedAt,
          processedCount: row.processedCount ?? 0,
          durationMs: completedAt - row.startedAt,
        },
      });
    }
    return null;
  },
});

export const getOpenRunForOrg = internalQuery({
  args: { organizationId: v.string() },
  returns: v.union(
    v.object({
      _id: v.id('retentionRuns'),
      startedAt: v.number(),
      lastCursor: v.optional(cursorValidator),
      processedCount: v.optional(v.number()),
    }),
    v.null(),
  ),
  handler: async (ctx, args) => {
    const recent = await ctx.db
      .query('retentionRuns')
      .withIndex('by_organizationId_startedAt', (q) =>
        q.eq('organizationId', args.organizationId),
      )
      .order('desc')
      .first();
    if (!recent || recent.completedAt !== undefined) return null;
    return {
      _id: recent._id,
      startedAt: recent.startedAt,
      lastCursor: recent.lastCursor,
      processedCount: recent.processedCount,
    };
  },
});

/**
 * Public — admin operator panel "longest-running cleanup" view.
 * Returns the most recent run for the org (in-flight or completed).
 */
export const getRetentionRunStatus = query({
  args: { organizationId: v.string() },
  handler: async (ctx, args) => {
    const authUser = await getAuthUserIdentity(ctx);
    if (!authUser) {
      throw new ConvexError({
        code: 'unauthenticated',
        message: 'Sign in required.',
      });
    }
    const member = await getOrganizationMember(
      ctx,
      args.organizationId,
      authUser,
    );
    if (!isAdmin(member.role)) {
      throw new ConvexError({
        code: 'forbidden',
        message: 'Admin role required to view retention run status.',
      });
    }

    const recent = await ctx.db
      .query('retentionRuns')
      .withIndex('by_organizationId_startedAt', (q) =>
        q.eq('organizationId', args.organizationId),
      )
      .order('desc')
      .take(5);
    return recent.map((r) => ({
      runId: r._id,
      startedAt: r.startedAt,
      completedAt: r.completedAt,
      lastCursor: r.lastCursor,
      lastError: r.lastError,
      processedCount: r.processedCount ?? 0,
      durationMs:
        r.completedAt !== undefined ? r.completedAt - r.startedAt : null,
    }));
  },
});
