feat: Litigation AI assistant and command center enhancements

migrations/0013_scrape_jobs.sql

@@ -0,0 +1,31 @@
+-- 0013_scrape_jobs.sql — Scrape job orchestration + identity binding
+
+-- Scrape job queue with retry, status tracking, and ChittyID binding
+CREATE TABLE IF NOT EXISTS cc_scrape_jobs (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  chitty_id VARCHAR(64),
+  job_type VARCHAR(50) NOT NULL,
+  target JSONB NOT NULL,
+  status VARCHAR(20) NOT NULL DEFAULT 'queued',
+  attempt INTEGER NOT NULL DEFAULT 0,
+  max_attempts INTEGER NOT NULL DEFAULT 3,
+  scheduled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  started_at TIMESTAMPTZ,
+  completed_at TIMESTAMPTZ,
+  result JSONB,
+  error_message TEXT,
+  parent_job_id UUID REFERENCES cc_scrape_jobs(id),
+  cron_source VARCHAR(30),
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_cc_scrape_jobs_status ON cc_scrape_jobs(status, scheduled_at);
+CREATE INDEX idx_cc_scrape_jobs_type ON cc_scrape_jobs(job_type);
+CREATE INDEX idx_cc_scrape_jobs_chitty ON cc_scrape_jobs(chitty_id);
+
+-- Add chitty_id to existing tables for identity binding
+ALTER TABLE cc_sync_log ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64);
+ALTER TABLE cc_legal_deadlines ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64);
+ALTER TABLE cc_properties ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64);
+ALTER TABLE cc_documents ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64);
+ALTER TABLE cc_obligations ADD COLUMN IF NOT EXISTS chitty_id VARCHAR(64);

src/db/schema.ts

@@ -1,4 +1,4 @@
-import { pgTable, uuid, text, numeric, boolean, integer, date, timestamp, jsonb, index } from 'drizzle-orm/pg-core';
+import { pgTable, uuid, varchar, text, numeric, boolean, integer, date, timestamp, jsonb, index } from 'drizzle-orm/pg-core';
 import { sql } from 'drizzle-orm';
 
 // ── Accounts ──────────────────────────────────────────────────
@@ -21,6 +21,7 @@ export const ccAccounts = pgTable('cc_accounts', {
 // ── Obligations ───────────────────────────────────────────────
 export const ccObligations = pgTable('cc_obligations', {
   id: uuid('id').primaryKey().defaultRandom(),
+  chittyId: varchar('chitty_id', { length: 64 }),
   accountId: uuid('account_id').references(() => ccAccounts.id),
   category: text('category').notNull(),
   subcategory: text('subcategory'),
@@ -73,6 +74,7 @@ export const ccTransactions = pgTable('cc_transactions', {
 // ── Properties ────────────────────────────────────────────────
 export const ccProperties = pgTable('cc_properties', {
   id: uuid('id').primaryKey().defaultRandom(),
+  chittyId: varchar('chitty_id', { length: 64 }),
   propertyName: text('property_name'),
   address: text('address').notNull(),
   unit: text('unit'),
@@ -93,6 +95,7 @@ export const ccProperties = pgTable('cc_properties', {
 // ── Legal Deadlines ───────────────────────────────────────────
 export const ccLegalDeadlines = pgTable('cc_legal_deadlines', {
   id: uuid('id').primaryKey().defaultRandom(),
+  chittyId: varchar('chitty_id', { length: 64 }),
   caseRef: text('case_ref').notNull(),
   caseSystem: text('case_system'),
   deadlineType: text('deadline_type').notNull(),
@@ -148,6 +151,7 @@ export const ccDisputeCorrespondence = pgTable('cc_dispute_correspondence', {
 // ── Documents ─────────────────────────────────────────────────
 export const ccDocuments = pgTable('cc_documents', {
   id: uuid('id').primaryKey().defaultRandom(),
+  chittyId: varchar('chitty_id', { length: 64 }),
   docType: text('doc_type').notNull(),
   source: text('source').notNull(),
   filename: text('filename'),
@@ -284,6 +288,7 @@ export const ccPaymentPlans = pgTable('cc_payment_plans', {
 // ── Sync Log ──────────────────────────────────────────────────
 export const ccSyncLog = pgTable('cc_sync_log', {
   id: uuid('id').primaryKey().defaultRandom(),
+  chittyId: varchar('chitty_id', { length: 64 }),
   source: text('source').notNull(),
   syncType: text('sync_type').notNull(),
   status: text('status').notNull(),
@@ -323,3 +328,26 @@ export const ccTasks = pgTable('cc_tasks', {
   priorityIdx: index('idx_cc_tasks_priority').on(table.priority),
   typeIdx: index('idx_cc_tasks_type').on(table.taskType),
 }));
+
+// ── Scrape Jobs ─────────────────────────────────────────────
+export const ccScrapeJobs = pgTable('cc_scrape_jobs', {
+  id: uuid('id').primaryKey().defaultRandom(),
+  chittyId: varchar('chitty_id', { length: 64 }),
+  jobType: varchar('job_type', { length: 50 }).notNull(),
+  target: jsonb('target').notNull(),
+  status: varchar('status', { length: 20 }).notNull().default('queued'),
+  attempt: integer('attempt').notNull().default(0),
+  maxAttempts: integer('max_attempts').notNull().default(3),
+  scheduledAt: timestamp('scheduled_at', { withTimezone: true }).defaultNow(),
+  startedAt: timestamp('started_at', { withTimezone: true }),
+  completedAt: timestamp('completed_at', { withTimezone: true }),
+  result: jsonb('result'),
+  errorMessage: text('error_message'),
+  parentJobId: uuid('parent_job_id'),
+  cronSource: varchar('cron_source', { length: 30 }),
+  createdAt: timestamp('created_at', { withTimezone: true }).defaultNow(),
+}, (table) => ({
+  statusIdx: index('idx_cc_scrape_jobs_status').on(table.status, table.scheduledAt),
+  typeIdx: index('idx_cc_scrape_jobs_type').on(table.jobType),
+  chittyIdx: index('idx_cc_scrape_jobs_chitty').on(table.chittyId),
+}));

src/index.ts

@@ -22,12 +22,14 @@ import { paymentPlanRoutes } from './routes/payment-plan';
 import { revenueRoutes } from './routes/revenue';
 import { emailConnectionRoutes } from './routes/email-connections';
 import { chatRoutes } from './routes/chat';
+import { litigationRoutes } from './routes/litigation';
 import { taskRoutes } from './routes/tasks';
 import { sendBeacon } from './lib/beacon';
 import { contextRoutes } from './routes/context';
 import { connectRoutes } from './routes/connect';
 import { ledgerRoutes } from './routes/ledger';
 import { tokenManagementRoutes } from './routes/token-management';
+import { jobRoutes } from './routes/jobs';
 
 export type Env = {
   AI: Ai;
@@ -121,6 +123,7 @@ app.route('/api/payment-plan', paymentPlanRoutes);
 app.route('/api/revenue', revenueRoutes);
 app.route('/api/email-connections', emailConnectionRoutes);
 app.route('/api/chat', chatRoutes);
+app.route('/api/litigation', litigationRoutes);
 app.route('/api/tasks', taskRoutes);
 // Identity (authenticated)
 app.route('/api/v1', metaRoutes);
@@ -132,6 +135,8 @@ app.route('/api/v1', connectRoutes);
 app.route('/api/v1', ledgerRoutes);
 // Token management (authenticated admin)
 app.route('/api/v1', tokenManagementRoutes);
+// Scrape job management (authenticated)
+app.route('/api/v1', jobRoutes);
 
 // MCP server — authenticated via shared token in KV
 app.use('/mcp/*', mcpAuthMiddleware);
@@ -141,7 +146,7 @@ export default {
   fetch: app.fetch,
   async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
     const sql = getDb(env);
-    ctx.waitUntil(runCronSync(event, env, sql));
+    ctx.waitUntil(runCronSync(event, env, sql, ctx));
     ctx.waitUntil(sendBeacon(env));
   },
 };

src/lib/cron.ts

@@ -7,6 +7,7 @@ import { generateProjections } from './projections';
 import { discoverRevenueSources } from './revenue';
 import { generatePaymentPlan, savePaymentPlan } from './payment-planner';
 import { reconcileNotionDisputes } from './dispute-sync';
+import { enqueueJob, processQueue } from './job-dispatcher';
 
 /**
  * Cron sync orchestrator.
@@ -18,6 +19,7 @@ export async function runCronSync(
   event: ScheduledEvent,
   env: Env,
   sql: NeonQueryFunction<false, false>,
+  ctx?: ExecutionContext,
 ): Promise<void> {
   const cronSources: Record<string, string> = {
     '0 12 * * *': 'daily_api',
@@ -136,15 +138,27 @@ export async function runCronSync(
     }
 
     if (source === 'utility_scrape') {
-      // Weekly utility portal scrapes via ChittyRouter
+      // Weekly utility portal scrapes via dispatcher
+      const chittyId = await env.COMMAND_KV.get('default:chitty_id') || undefined;
       const utilityTargets = ['comed', 'peoples_gas', 'xfinity'];
       for (const target of utilityTargets) {
         try {
-          recordsSynced += await syncPortal(env, sql, target);
+          await enqueueJob(sql, 'portal_scrape', { portal: target }, {
+            chittyId,
+            cronSource: 'utility_scrape',
+          });
         } catch (err) {
-          console.error(`[cron:utility:${target}] failed:`, err);
+          console.error(`[cron:utility:${target}] enqueue failed:`, err);
         }
       }
+      // Process all queued utility jobs
+      try {
+        const queueResult = await processQueue(sql, env, ctx);
+        recordsSynced += queueResult.succeeded;
+        console.log(`[cron:utility] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`);
+      } catch (err) {
+        console.error('[cron:utility] processQueue failed:', err);
+      }
 
       // Also pull email-parsed bills (also called in daily_api — upsert prevents duplicates)
       try {
@@ -156,15 +170,23 @@ export async function runCronSync(
 
     if (source === 'court_docket') {
       try {
-        recordsSynced += await syncCourtDocket(env, sql);
+        // Enqueue via dispatcher for retry + fan-out
+        const chittyId = await env.COMMAND_KV.get('default:chitty_id') || undefined;
+        await enqueueJob(sql, 'court_docket', { case_number: '2024D007847' }, {
+          chittyId,
+          cronSource: 'court_docket',
+        });
+        const queueResult = await processQueue(sql, env, ctx);
+        recordsSynced += queueResult.succeeded;
+        console.log(`[cron:court_docket] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`);
       } catch (err) {
         console.error('[cron:court_docket] failed:', err);
       }
     }
 
     if (source === 'monthly_check') {
       try {
-        recordsSynced += await syncMonthlyChecks(env, sql);
+        recordsSynced += await syncMonthlyChecksViaDispatcher(env, sql, ctx);
       } catch (err) {
         console.error('[cron:monthly_check] failed:', err);
       }
@@ -544,6 +566,48 @@ async function syncMonthlyChecks(env: Env, sql: NeonQueryFunction<false, false>)
   return synced;
 }
 
+/**
+ * Monthly scrapers via dispatcher — enqueues Mr. Cooper + all property tax PINs as jobs.
+ */
+async function syncMonthlyChecksViaDispatcher(
+  env: Env,
+  sql: NeonQueryFunction<false, false>,
+  ctx?: ExecutionContext,
+): Promise<number> {
+  const chittyId = await env.COMMAND_KV.get('default:chitty_id') || undefined;
+
+  // Enqueue Mr. Cooper
+  try {
+    await enqueueJob(sql, 'mr_cooper', { property: 'addison' }, {
+      chittyId,
+      cronSource: 'monthly_check',
+    });
+  } catch (err) {
+    console.error('[cron:mr_cooper] enqueue failed:', err);
+  }
+
+  // Enqueue Cook County tax for each property with a PIN
+  try {
+    const properties = await sql`SELECT id, tax_pin FROM cc_properties WHERE tax_pin IS NOT NULL`;
+    for (const prop of properties) {
+      await enqueueJob(sql, 'cook_county_tax', {
+        pin: prop.tax_pin as string,
+        property_id: prop.id as string,
+      }, {
+        chittyId,
+        cronSource: 'monthly_check',
+      });
+    }
+  } catch (err) {
+    console.error('[cron:cook_county_tax] enqueue failed:', err);
+  }
+
+  // Process all enqueued monthly jobs
+  const queueResult = await processQueue(sql, env, ctx);
+  console.log(`[cron:monthly] dispatcher: ${queueResult.succeeded} succeeded, ${queueResult.failed} failed`);
+  return queueResult.succeeded;
+}
+
 /**
  * Sync a bill portal via ChittyRouter gateway.
  * ChittyRouter fetches credentials from ChittyConnect, dispatches to ChittyScrape,