Move queue state out of analysis JSON into real columns

migrations/007_apps_state_columns.sql

@@ -0,0 +1,69 @@
+-- Move queue/analysis state out of the analysis JSON payload into real columns.
+-- The analysis column keeps the raw analyser payload untouched (so the website's
+-- failure display and the Raspberry Pi HTTP contract stay byte-identical);
+-- scheduling state now lives in status/processing_started/failure_* columns.
+--
+-- Additive and backward-compatible: the currently deployed code ignores these
+-- columns entirely, and the defaults keep new rows valid until the new code boots.
+
+ALTER TABLE apps
+    ADD COLUMN status text NOT NULL DEFAULT 'queued'
+        CHECK (status IN ('queued', 'processing', 'analysed', 'failed')),
+    ADD COLUMN processing_started timestamptz,
+    ADD COLUMN failure_reason text,
+    ADD COLUMN failure_retryable boolean;
+
+-- Backfill from the JSON state previously embedded in analysis.
+-- Order matters: the in-flight processing indicator also carries
+-- success=false, so it must be classified before the failed branch.
+--
+-- Timestamps are validated with a regex before casting so a malformed
+-- analysis->>'timestamp' cannot abort the whole migration; unparseable
+-- values fall back to NOW() so the processing lock still expires normally.
+UPDATE apps
+SET
+    status = CASE
+        WHEN analysis IS NULL THEN 'queued'
+        WHEN analysis->>'logs' = 'Processing in progress' THEN 'processing'
+        WHEN analysis->>'success' = 'false' THEN 'failed'
+        ELSE 'analysed'
+    END,
+    processing_started = CASE
+        WHEN analysis->>'logs' = 'Processing in progress' THEN
+            CASE
+                WHEN analysis->>'timestamp' ~ '^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}'
+                    THEN (analysis->>'timestamp')::timestamptz
+                ELSE NOW()
+            END
+    END,
+    failure_reason = CASE
+        WHEN COALESCE(analysis->>'logs', '') <> 'Processing in progress'
+            AND analysis->>'success' = 'false'
+            THEN analysis->>'reason'
+    END,
+    failure_retryable = CASE
+        WHEN COALESCE(analysis->>'logs', '') <> 'Processing in progress'
+            AND analysis->>'success' = 'false'
+            THEN COALESCE(analysis->>'retryable', 'true') <> 'false'
+    END;
+
+-- Indexes supporting the queue scan (models/Apps.js nextApp / countQueue).
+-- A partial index per hot status keeps the candidate lookup cheap.
+CREATE INDEX IF NOT EXISTS apps_status_idx
+    ON apps (status);
+
+CREATE INDEX IF NOT EXISTS apps_queued_added_idx
+    ON apps (added ASC)
+    WHERE status = 'queued';
+
+CREATE INDEX IF NOT EXISTS apps_processing_started_idx
+    ON apps (processing_started)
+    WHERE status = 'processing';
+
+CREATE INDEX IF NOT EXISTS apps_failed_retryable_idx
+    ON apps (analysisversion, analysed)
+    WHERE status = 'failed' AND failure_retryable;
+
+CREATE INDEX IF NOT EXISTS apps_analysed_version_idx
+    ON apps (analysisversion, analysed)
+    WHERE status = 'analysed';

migrations/008_jsonb_timestamptz.sql

@@ -0,0 +1,14 @@
+-- Modernise column types: json -> jsonb and timestamp -> timestamptz.
+-- Historical timestamps were written in UTC, so reinterpret them as UTC.
+-- schema_migrations is deliberately left untouched.
+
+ALTER TABLE apps
+    ALTER COLUMN details TYPE jsonb USING details::jsonb,
+    ALTER COLUMN analysis TYPE jsonb USING analysis::jsonb,
+    ALTER COLUMN added TYPE timestamptz USING (added AT TIME ZONE 'UTC'),
+    ALTER COLUMN analysed TYPE timestamptz USING (analysed AT TIME ZONE 'UTC');
+
+ALTER TABLE app_analyses
+    ALTER COLUMN analysis TYPE jsonb USING analysis::jsonb,
+    ALTER COLUMN analysed TYPE timestamptz USING (analysed AT TIME ZONE 'UTC'),
+    ALTER COLUMN app_store_updated TYPE timestamptz USING (app_store_updated AT TIME ZONE 'UTC');

migrations/009_case_insensitive_appids.sql

@@ -0,0 +1,55 @@
+-- The App Store treats bundle IDs case-insensitively; enforce that in the
+-- database. Merge rows whose appid differs only by case, then add a unique
+-- index on lower(appid).
+
+-- Rank rows within each case-insensitive duplicate group. The keeper (rn = 1)
+-- prefers status = 'analysed', then the most recent analysis, then the
+-- earliest-added row; appid is a final tie-breaker for determinism.
+CREATE TEMP TABLE apps_case_dupes ON COMMIT DROP AS
+SELECT
+    appid,
+    lower(appid) AS lower_appid,
+    row_number() OVER (
+        PARTITION BY lower(appid)
+        ORDER BY
+            (status = 'analysed') DESC,
+            analysed DESC NULLS LAST,
+            added ASC,
+            appid ASC
+    ) AS rn
+FROM apps
+WHERE lower(appid) IN (
+    SELECT lower(appid)
+    FROM apps
+    GROUP BY lower(appid)
+    HAVING count(*) > 1
+);
+
+-- Before repointing history rows to the keeper, drop rows that would collide
+-- on the (appid, analysed) unique index once every appid in the group becomes
+-- the keeper's: keep the lowest id per (group, analysed) pair.
+DELETE FROM app_analyses a
+USING apps_case_dupes ma, app_analyses b, apps_case_dupes mb
+WHERE ma.appid = a.appid
+    AND mb.appid = b.appid
+    AND ma.lower_appid = mb.lower_appid
+    AND a.analysed = b.analysed
+    AND a.id > b.id;
+
+-- Repoint the losers' remaining history rows to the keeper.
+UPDATE app_analyses a
+SET appid = keeper.appid
+FROM apps_case_dupes loser
+JOIN apps_case_dupes keeper
+    ON keeper.lower_appid = loser.lower_appid AND keeper.rn = 1
+WHERE a.appid = loser.appid
+    AND loser.rn > 1;
+
+-- Drop the loser apps rows (their history has been moved to the keeper).
+DELETE FROM apps a
+USING apps_case_dupes loser
+WHERE loser.rn > 1
+    AND a.appid = loser.appid;
+
+CREATE UNIQUE INDEX IF NOT EXISTS apps_appid_lower_unique
+    ON apps (lower(appid));

models/Apps.js

@@ -14,8 +14,27 @@ pool.on('error', (err) => {
     console.error('Unexpected PostgreSQL pool error:', err.message);
 });
 
+// Derive the scheduling-state columns from an analyser payload.
+// A payload with success === false is a failure; everything else is a
+// successful analysis. Kept as a pure, exported helper so the mapping can be
+// unit-tested without a database.
+function deriveAnalysisState(analysis) {
+    if (analysis && analysis.success === false) {
+        return {
+            status: 'failed',
+            failureReason: (analysis.reason || analysis.logs) || null,
+            failureRetryable: analysis.retryable !== false
+        };
+    }
+    return {
+        status: 'analysed',
+        failureReason: null,
+        failureRetryable: null
+    };
+}
+
 const lastAnalysed = async () => {
-    const result = await pool.query('SELECT * FROM apps WHERE analysis IS NOT NULL ORDER BY analysed DESC LIMIT 5');
+    const result = await pool.query("SELECT * FROM apps WHERE status = 'analysed' ORDER BY analysed DESC LIMIT 5");
     return result.rows;
 }
 
@@ -26,7 +45,7 @@ const healthCheck = async () => {
 const findApp = async (appId) => {
     if (!isValidAppId(appId)) return null;
 
-    const result = await pool.query('SELECT * FROM apps WHERE appid = $1', [appId]);
+    const result = await pool.query('SELECT * FROM apps WHERE lower(appid) = lower($1)', [appId]);
     if (result.rows.length == 0)
         return null;
 
@@ -38,7 +57,7 @@ const countQueue = async (added) => {
         const result = await pool.query(`
             SELECT COUNT(*)
             FROM apps
-            WHERE analysis IS NULL
+            WHERE status = 'queued'
                 AND added < $1
                 AND appid ~ $2
                 AND appid !~ '[.]$'
@@ -49,7 +68,7 @@ const countQueue = async (added) => {
         const result = await pool.query(`
             SELECT COUNT(*)
             FROM apps
-            WHERE analysis IS NULL
+            WHERE status = 'queued'
                 AND appid ~ $1
                 AND appid !~ '[.]$'
                 AND length(appid) <= $2
@@ -62,7 +81,9 @@ const addApp = async (appId, details) => {
     if (!isValidAppId(appId)) throw new TypeError('Invalid App Store bundle ID');
     if (!details || details.appId !== appId) throw new TypeError('App Store bundle ID mismatch');
 
-    const result = await pool.query('INSERT INTO apps (appid, details) VALUES ($1, $2) ON CONFLICT (appid) DO NOTHING', [appId, details]);
+    // Bare ON CONFLICT covers both the appid primary key and the
+    // case-insensitive lower(appid) unique index from migration 009.
+    const result = await pool.query('INSERT INTO apps (appid, details) VALUES ($1, $2) ON CONFLICT DO NOTHING', [appId, details]);
     return result;
 }
 
@@ -76,66 +97,39 @@ const currentAnalysisVersion = parseInt(process.env.CURRENT_ANALYSIS_VERSION ||
 const staleAnalysisDays = parseInt(process.env.STALE_ANALYSIS_DAYS || '180', 10);
 const processingTimeoutMinutes = parseInt(process.env.PROCESSING_TIMEOUT_MINUTES || '120', 10);
 
-async function snapshotCurrentAnalysis(client, appId) {
-    await client.query(`
-        INSERT INTO app_analyses (
-            appid,
-            analysis,
-            analysisversion,
-            analysed,
-            app_version,
-            app_store_updated,
-            analysis_source,
-            success
-        )
-        SELECT
-            appid,
-            analysis,
-            analysisversion,
-            COALESCE(analysed, NOW()),
-            details->>'version',
-            NULLIF(details->>'updated', '')::timestamp,
-            COALESCE(analysis->>'analysis_source', 'legacy'),
-            COALESCE((analysis->>'success')::boolean, true)
-        FROM apps
-        WHERE appid = $1
-            AND analysis IS NOT NULL
-            AND NOT EXISTS (
-                SELECT 1
-                FROM app_analyses existing
-                WHERE existing.appid = apps.appid
-                    AND existing.analysed = COALESCE(apps.analysed, NOW())
-            )
-        ON CONFLICT (appid, analysed) DO NOTHING
-    `, [appId]);
-}
-
 const nextApp = async () => {
-    const processingIndicator = {
-        success: false,
-        logs: 'Processing in progress',
-        timestamp: new Date().toISOString() // ISO 8601 format
-    };
-
     const client = await pool.connect();
     try {
         await client.query('BEGIN');
 
+        // Candidate selection keys off the real status columns instead of the
+        // analysis JSON. The lock is non-destructive: we flip status to
+        // 'processing' and stamp processing_started, but leave the analysis
+        // payload alone so the last good result stays visible on the website
+        // while a refetch is in flight. History snapshots happen in
+        // updateAnalysis when the new result lands.
         const candidate = await client.query(`
-            SELECT appid, analysis
+            SELECT appid
             FROM apps
             WHERE appid ~ $4
                 AND appid !~ '[.]$'
                 AND length(appid) <= $5
-                AND COALESCE(analysis->>'retryable', 'true') <> 'false'
                 AND (
-                    analysis IS NULL
+                    status = 'queued'
+                    OR (
+                        status = 'processing'
+                        AND processing_started < NOW() - ($3::int * INTERVAL '1 minute')
+                    )
                     OR (
-                        analysis->>'logs' = 'Processing in progress'
-                        AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute')
+                        status = 'analysed'
+                        AND (
+                            analysisversion IS DISTINCT FROM $1
+                            OR analysed < NOW() - ($2::int * INTERVAL '1 day')
+                        )
                     )
                     OR (
-                        coalesce(analysis->>'logs', '') <> 'Processing in progress'
+                        status = 'failed'
+                        AND failure_retryable
                         AND (
                             analysisversion IS DISTINCT FROM $1
                             OR analysed < NOW() - ($2::int * INTERVAL '1 day')
@@ -158,17 +152,13 @@ const nextApp = async () => {
             return null;
         }
 
-        const app = candidate.rows[0];
-        if (app.analysis && app.analysis.logs !== 'Processing in progress') {
-            await snapshotCurrentAnalysis(client, app.appid);
-        }
-
         const result = await client.query(`
             UPDATE apps
-            SET analysis = $1
-            WHERE appid = $2
+            SET status = 'processing',
+                processing_started = NOW()
+            WHERE appid = $1
             RETURNING appid
-        `, [JSON.stringify(processingIndicator), app.appid]);
+        `, [candidate.rows[0].appid]);
 
         await client.query('COMMIT');
 
@@ -185,13 +175,28 @@ const nextApp = async () => {
 const updateAnalysis = async (appId, analysis, analysisVersion) => {
     if (!isValidAppId(appId)) throw new TypeError('Invalid App Store bundle ID');
 
+    const { status, failureReason, failureRetryable } = deriveAnalysisState(analysis);
+
     const client = await pool.connect();
     try {
         await client.query('BEGIN');
 
+        // Keep writing the raw payload to apps.analysis (the website failure
+        // display and existing HTTP responses depend on it) AND set the new
+        // scheduling columns. processing_started is cleared now that the lock
+        // is resolved.
         const result = await client.query(
-            'UPDATE apps SET analysis = $1, analysisVersion = $2, analysed = NOW() WHERE appid = $3 RETURNING appid, details, analysed',
-            [analysis, analysisVersion, appId]
+            `UPDATE apps
+             SET analysis = $1,
+                 analysisVersion = $2,
+                 analysed = NOW(),
+                 status = $4,
+                 failure_reason = $5,
+                 failure_retryable = $6,
+                 processing_started = NULL
+             WHERE appid = $3
+             RETURNING appid, details, analysed`,
+            [analysis, analysisVersion, appId, status, failureReason, failureRetryable]
         );
 
         if (result.rowCount > 0) {
@@ -241,21 +246,19 @@ const updateAnalysis = async (appId, analysis, analysisVersion) => {
 }
 
 const getAllApps = async () => {
-    const result = await pool.query('SELECT * FROM apps WHERE analysis IS NOT NULL');
+    const result = await pool.query("SELECT * FROM apps WHERE status = 'analysed'");
     return result.rows;
 }
 
 const getSiteDataSignature = async () => {
     const result = await pool.query(`
         SELECT
             COUNT(*) FILTER (
-                WHERE analysis IS NOT NULL
-                    AND COALESCE(analysis->>'success', 'true') != 'false'
+                WHERE status = 'analysed'
                     AND analysis->'trackers' IS NOT NULL
             ) AS app_count,
             MAX(analysed) FILTER (
-                WHERE analysis IS NOT NULL
-                    AND COALESCE(analysis->>'success', 'true') != 'false'
+                WHERE status = 'analysed'
                     AND analysis->'trackers' IS NOT NULL
             ) AS latest_analysis
         FROM apps
@@ -269,7 +272,7 @@ const getSiteDataSignature = async () => {
 }
 
 const countAnalysed = async () => {
-    const result = await pool.query("SELECT COUNT(*) FROM apps WHERE analysis IS NOT NULL AND COALESCE(analysis->>'success', 'true') != 'false'");
+    const result = await pool.query("SELECT COUNT(*) FROM apps WHERE status = 'analysed'");
     return parseInt(result.rows[0].count, 10);
 }
 
@@ -283,5 +286,6 @@ module.exports = {
     updateAnalysis,
     getAllApps,
     getSiteDataSignature,
-    healthCheck
+    healthCheck,
+    deriveAnalysisState
 }