diff --git a/bot.js b/bot.js index 7f674af..987ea4f 100644 --- a/bot.js +++ b/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,6 +3270,12 @@ function handleNewAudio(audioData) { }); }); } + if (transcriptionJobId) { + safelyUpdateProcessingJob('mark transcription job completed', () => markJobCompleted(db, transcriptionJobId, { + empty: false, + transcriptionLength: transcriptionText.length + })); + } logger.info(`Successfully processed: ${filename}`); }); }; @@ -3241,6 +3283,9 @@ function handleNewAudio(audioData) { // --- 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); diff --git a/src/db/migrations.js b/src/db/migrations.js index 45e97b0..36e519a 100644 --- a/src/db/migrations.js +++ b/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)` + ] } ]; diff --git a/src/jobs/processingJobs.js b/src/jobs/processingJobs.js new file mode 100644 index 0000000..345fb1e --- /dev/null +++ b/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 +}; diff --git a/test/migrations.test.js b/test/migrations.test.js index 5bbe56b..3c1fd3e 100644 --- a/test/migrations.test.js +++ b/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'] ); }); diff --git a/test/processingJobs.test.js b/test/processingJobs.test.js new file mode 100644 index 0000000..14d1984 --- /dev/null +++ b/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 }); +});