Add persistent processing job records

bot.js

@@ -4,6 +4,13 @@ require('dotenv').config();
 const { loadConfig } = require('./src/config');
 const { applyMigrations } = require('./src/db/migrations');
 const { normalizeIncomingCall } = require('./src/ingestion/normalizeCall');
+const {
+  JOB_TYPES,
+  createProcessingJob,
+  markJobCompleted,
+  markJobFailed,
+  markJobProcessing
+} = require('./src/jobs/processingJobs');
 
 // Get environment variables first, before any usage
 const {
@@ -2901,6 +2908,12 @@ function handleNewAudio(audioData) {
     phase2_tdma,
     color_code
   } = audioData;
+
+  const safelyUpdateProcessingJob = (actionDescription, updateFn) => {
+    updateFn().catch((jobError) => {
+      logger.warn(`Could not ${actionDescription}: ${jobError.message}`);
+    });
+  };
 
   // Log srcList for debugging
   logger.info(`[handleNewAudio] Received audio data for ${filename}, srcList=${srcList ? (typeof srcList === 'string' ? `string(${srcList.substring(0, 100)}...)` : `object`) : 'null/undefined'}, isTrunkRecorder=${isTrunkRecorder}`);
@@ -3011,7 +3024,7 @@ function handleNewAudio(audioData) {
     db.run(
       `INSERT INTO transcriptions (talk_group_id, timestamp, transcription, audio_file_path, address, lat, lon) VALUES (?, ?, ?, ?, NULL, NULL, NULL)`,
       [talkGroupID, unixTimestampSeconds, '', storagePath], // Use the Unix timestamp
-      function (err) {
+      async function (err) {
         if (err) {
           logger.error(`Error inserting initial transcription record for ${filename}:`, err);
           // If DB insert fails, delete the temp file
@@ -3022,8 +3035,25 @@ function handleNewAudio(audioData) {
         }
 
         const transcriptionId = this.lastID; // Get the ID from the database insert
+        let transcriptionJobId = null;
         logger.info(`Created transcription record ID ${transcriptionId} using storage path: ${storagePath}`);
 
+        try {
+          transcriptionJobId = await createProcessingJob(db, {
+            transcriptionId,
+            jobType: JOB_TYPES.TRANSCRIPTION,
+            payload: {
+              filename,
+              talkGroupID,
+              transcriptionMode: effectiveTranscriptionMode,
+              storageMode: STORAGE_MODE
+            }
+          });
+          logger.info(`Created transcription job ${transcriptionJobId} for transcription ID ${transcriptionId}`);
+        } catch (jobError) {
+          logger.warn(`Could not create transcription job for ID ${transcriptionId}: ${jobError.message}`);
+        }
+
         // Conditionally insert audio blob for Listen Live feature (local storage only)
         if (STORAGE_MODE !== 's3') {
             db.run(
@@ -3164,6 +3194,12 @@ function handleNewAudio(audioData) {
                   logger.warn(warningMsg);
                   updateTranscription(transcriptionId, "", async () => {
                     logger.info(`Updated DB with empty transcription for ID ${transcriptionId}`);
+                    if (transcriptionJobId) {
+                      safelyUpdateProcessingJob('mark transcription job completed', () => markJobCompleted(db, transcriptionJobId, {
+                        empty: true,
+                        reason: 'no_transcription'
+                      }));
+                    }
 
                     // *** IMPORTANT: Check for two-tone even with empty transcription ***
                     // Tone files might contain only tones without voice content
@@ -3234,13 +3270,22 @@ function handleNewAudio(audioData) {
                       });
                     });
                   }
+                  if (transcriptionJobId) {
+                    safelyUpdateProcessingJob('mark transcription job completed', () => markJobCompleted(db, transcriptionJobId, {
+                      empty: false,
+                      transcriptionLength: transcriptionText.length
+                    }));
+                  }
                   logger.info(`Successfully processed: ${filename}`);
                 });
             };
             // --- End common callback definition ---
 
             // --- Choose transcription method based on mode ---
             logger.info(`Initiating transcription for ID ${transcriptionId} using mode: ${effectiveTranscriptionMode}`);
+            if (transcriptionJobId) {
+              safelyUpdateProcessingJob('mark transcription job processing', () => markJobProcessing(db, transcriptionJobId));
+            }
 
             if (effectiveTranscriptionMode === 'openai') {
                 // OpenAI API transcription mode
@@ -3373,6 +3418,9 @@ function handleNewAudio(audioData) {
               logger.error(`Error uploading audio to S3 for transcription ID ${transcriptionId} (key: ${storagePath}):`, s3Err);
               }
               // If S3 upload fails, should we delete the DB record?
+              if (transcriptionJobId) {
+                safelyUpdateProcessingJob('mark transcription job failed', () => markJobFailed(db, transcriptionJobId, s3Err));
+              }
               db.run('DELETE FROM transcriptions WHERE id = ?', [transcriptionId], () => {});
               fs.unlink(tempPath, (errUnlink) => { // Delete temp file on S3 error
                   if (errUnlink) logger.error(`Error deleting temp file after S3 upload error ${tempPath}:`, errUnlink);
@@ -3390,6 +3438,9 @@ function handleNewAudio(audioData) {
               if (renameErr) {
                   logger.error(`Error moving temp file ${tempPath} to final location ${finalLocalPath}:`, renameErr);
                   // If rename fails, delete DB record and original temp file
+                  if (transcriptionJobId) {
+                    safelyUpdateProcessingJob('mark transcription job failed', () => markJobFailed(db, transcriptionJobId, renameErr));
+                  }
                   db.run('DELETE FROM transcriptions WHERE id = ?', [transcriptionId], () => {});
                   fs.unlink(tempPath, (errUnlink) => {
                       if (errUnlink) logger.error(`Error deleting temp file after rename error ${tempPath}:`, errUnlink);

src/db/migrations.js

@@ -36,7 +36,7 @@ const BASE_MIGRATIONS = [
         transcription_id INTEGER,
         audio_data BLOB,
         created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
-        FOREIGN KEY(transcription_id) REFERENCES transcriptions(id)
+        FOREIGN KEY(transcription_id) REFERENCES transcriptions(id) ON DELETE SET NULL
       )`
     ]
   },
@@ -64,6 +64,31 @@ const BASE_MIGRATIONS = [
         FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
       )`
     ]
+  },
+  {
+    id: '003_create_call_jobs',
+    statements: [
+      `CREATE TABLE IF NOT EXISTS call_jobs (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        transcription_id INTEGER,
+        job_type TEXT NOT NULL,
+        status TEXT NOT NULL DEFAULT 'pending',
+        attempts INTEGER NOT NULL DEFAULT 0,
+        max_attempts INTEGER NOT NULL DEFAULT 3,
+        priority INTEGER NOT NULL DEFAULT 0,
+        run_after DATETIME,
+        payload_json TEXT,
+        result_json TEXT,
+        last_error TEXT,
+        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+        started_at DATETIME,
+        completed_at DATETIME,
+        FOREIGN KEY(transcription_id) REFERENCES transcriptions(id)
+      )`,
+      `CREATE INDEX IF NOT EXISTS idx_call_jobs_status_priority ON call_jobs (status, priority DESC, created_at ASC)`,
+      `CREATE INDEX IF NOT EXISTS idx_call_jobs_transcription_type ON call_jobs (transcription_id, job_type)`
+    ]
   }
 ];
 

src/jobs/processingJobs.js

@@ -0,0 +1,129 @@
+const JOB_TYPES = {
+  TRANSCRIPTION: 'transcription',
+  ADDRESS_EXTRACTION: 'address_extraction',
+  GEOCODING: 'geocoding',
+  DISCORD_PUBLISH: 'discord_publish'
+};
+
+const JOB_STATUS = {
+  PENDING: 'pending',
+  PROCESSING: 'processing',
+  COMPLETED: 'completed',
+  FAILED: 'failed',
+  RETRYABLE: 'retryable'
+};
+
+function serializeJson(value) {
+  if (value === undefined) return null;
+  return JSON.stringify(value);
+}
+
+function parseJson(value, fallback = null) {
+  if (!value) return fallback;
+  try {
+    return JSON.parse(value);
+  } catch {
+    return fallback;
+  }
+}
+
+function run(db, sql, params = []) {
+  return new Promise((resolve, reject) => {
+    db.run(sql, params, function onRun(err) {
+      if (err) reject(err);
+      else resolve(this);
+    });
+  });
+}
+
+function get(db, sql, params = []) {
+  return new Promise((resolve, reject) => {
+    db.get(sql, params, (err, row) => {
+      if (err) reject(err);
+      else resolve(row);
+    });
+  });
+}
+
+async function createProcessingJob(db, {
+  transcriptionId,
+  jobType,
+  payload = {},
+  priority = 0,
+  maxAttempts = 3,
+  runAfter = null
+}) {
+  const result = await run(
+    db,
+    `INSERT INTO call_jobs (
+      transcription_id, job_type, status, priority, max_attempts, run_after, payload_json
+    ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+    [
+      transcriptionId,
+      jobType,
+      JOB_STATUS.PENDING,
+      priority,
+      maxAttempts,
+      runAfter,
+      serializeJson(payload)
+    ]
+  );
+
+  return result.lastID;
+}
+
+async function markJobProcessing(db, jobId) {
+  await run(
+    db,
+    `UPDATE call_jobs
+     SET status = ?, attempts = attempts + 1, started_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
+     WHERE id = ?`,
+    [JOB_STATUS.PROCESSING, jobId]
+  );
+}
+
+async function markJobCompleted(db, jobId, result = {}) {
+  await run(
+    db,
+    `UPDATE call_jobs
+     SET status = ?, result_json = ?, completed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
+     WHERE id = ?`,
+    [JOB_STATUS.COMPLETED, serializeJson(result), jobId]
+  );
+}
+
+async function markJobFailed(db, jobId, error, { retryable = false } = {}) {
+  const status = retryable ? JOB_STATUS.RETRYABLE : JOB_STATUS.FAILED;
+  const message = error instanceof Error ? error.message : String(error || 'Unknown error');
+
+  await run(
+    db,
+    `UPDATE call_jobs
+     SET status = ?, last_error = ?, completed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
+     WHERE id = ?`,
+    [status, message, jobId]
+  );
+}
+
+async function getJobById(db, jobId) {
+  const row = await get(db, 'SELECT * FROM call_jobs WHERE id = ?', [jobId]);
+  if (!row) return null;
+
+  return {
+    ...row,
+    payload: parseJson(row.payload_json, {}),
+    result: parseJson(row.result_json, null)
+  };
+}
+
+module.exports = {
+  JOB_STATUS,
+  JOB_TYPES,
+  createProcessingJob,
+  getJobById,
+  markJobCompleted,
+  markJobFailed,
+  markJobProcessing,
+  parseJson,
+  serializeJson
+};

test/migrations.test.js

@@ -6,13 +6,13 @@ const { getMigrationPlan } = require('../src/db/migrations');
 test('migration plan includes core tables by default', () => {
   assert.deepEqual(
     getMigrationPlan({ enableAuth: false }).map((migration) => migration.id),
-    ['001_create_core_tables']
+    ['001_create_core_tables', '003_create_call_jobs']
   );
 });
 
 test('migration plan includes auth tables when auth is enabled', () => {
   assert.deepEqual(
     getMigrationPlan({ enableAuth: true }).map((migration) => migration.id),
-    ['001_create_core_tables', '002_create_auth_tables']
+    ['001_create_core_tables', '002_create_auth_tables', '003_create_call_jobs']
   );
 });

test/processingJobs.test.js

@@ -0,0 +1,25 @@
+const test = require('node:test');
+const assert = require('node:assert/strict');
+
+const {
+  JOB_STATUS,
+  JOB_TYPES,
+  parseJson,
+  serializeJson
+} = require('../src/jobs/processingJobs');
+
+test('job constants define the first durable processing states', () => {
+  assert.equal(JOB_TYPES.TRANSCRIPTION, 'transcription');
+  assert.equal(JOB_STATUS.PENDING, 'pending');
+  assert.equal(JOB_STATUS.PROCESSING, 'processing');
+  assert.equal(JOB_STATUS.COMPLETED, 'completed');
+});
+
+test('serializeJson and parseJson preserve payload objects', () => {
+  const payload = { transcriptionId: 42, mode: 'local' };
+  assert.deepEqual(parseJson(serializeJson(payload)), payload);
+});
+
+test('parseJson returns fallback for invalid JSON', () => {
+  assert.deepEqual(parseJson('{bad json', { ok: false }), { ok: false });
+});