feat: watermark state-tracking with deterministic prioritized task queuing

package.json

@@ -17,6 +17,8 @@
     "refresh-test": "node scripts/refresh-test-data.js"
   },
   "dependencies": {
+    "@actions/cache": "^6.0.0",
+    "@actions/core": "^3.0.0",
     "@octokit/graphql": "^9.0.0",
     "@octokit/rest": "^22.0.0",
     "ajv": "^8.12.0",

src/github/api.js

@@ -366,9 +366,12 @@ async function getRecentItems(org, repos, monitoredUser, windowHours = undefined
     hours = Number.isFinite(windowHours) && windowHours > 0 ? windowHours : 24;
   }
 
+  // Use explicit since date if provided, otherwise slide based on hours
+  const sinceDateStr = options.since || new Date(Date.now() - hours * 60 * 60 * 1000).toISOString();
+
   // Backfill mode: BACKFILL=<org>/<repo> searches only that repo without a date filter
   const backfillRepo = process.env.BACKFILL && process.env.BACKFILL.includes('/') ? process.env.BACKFILL : null;
-  const sinceClause = backfillRepo ? '' : ` updated:>${new Date(Date.now() - hours * 60 * 60 * 1000).toISOString()}`;
+  const sinceClause = backfillRepo ? '' : ` updated:>=${sinceDateStr}`;
 
   if (backfillRepo) {
     logger.info(`🔄 BACKFILL mode — searching only ${backfillRepo} without date filter`);
@@ -394,7 +397,7 @@ async function getRecentItems(org, repos, monitoredUser, windowHours = undefined
 
   // Search for PRs authored by monitored user in allowed organizations only
   // Author/assignee searches always use the date filter (only repo search is unlimited in backfill)
-  const sinceFilter = ` updated:>${new Date(Date.now() - hours * 60 * 60 * 1000).toISOString()}`;
+  const sinceFilter = ` updated:>=${sinceDateStr}`;
   const authorOrgsQuery = allowedOrgs.map(o => `org:${o}`).join(' ');
   const authorSearchQuery = authorOrgsQuery
     ? `${authorOrgsQuery} author:${monitoredUser}${sinceFilter}`

src/index.js

@@ -63,6 +63,7 @@ import { EnvironmentValidator } from './utils/environment-validator.js';
 import { loadBoardRules } from './config/board-rules.js';
 import { loadEventItems } from './utils/event-items.js';
 import { auditLog } from './utils/audit-logger.js';
+import { getWatermark, saveWatermark } from './utils/watermark.js';
 
 
 // Custom error classes for robust error handling
@@ -238,6 +239,14 @@ async function main() {
       ? 1
       : undefined;
 
+    // Load gapless sync watermark from cache
+    const watermark = await getWatermark(context.projectId);
+    if (watermark) {
+      log.info(`Using gapless sync watermark: ${watermark}`);
+    } else {
+      log.info(`No watermark found, falling back to ${windowHours || process.env.UPDATE_WINDOW_HOURS || 24}h sliding window`);
+    }
+
     // Process items according to our enhanced rules
     const eventItems = await loadEventItems(eventName, process.env.GITHUB_EVENT_PATH);
     if (eventItems.length > 0) {
@@ -255,7 +264,8 @@ async function main() {
       projectId: context.projectId,
       windowHours,
       seedItems: eventItems,
-      allowedOrgs: context.allowedOrgs
+      allowedOrgs: context.allowedOrgs,
+      since: watermark
     });
 
     // Process additional rules for added items
@@ -469,6 +479,11 @@ async function main() {
       log.error(`Project Board Sync completed with ${errors.length} errors`);
     } else {
       log.info('Project Board Sync completed successfully');
+
+      // Save new watermark if we completed without CRITICAL errors.
+      // We use the start time of the run as the new watermark.
+      // This ensures that even if the run takes 10 minutes, the next run covers everything since this run started.
+      await saveWatermark(context.projectId, startTime.toISOString());
     }
 
     // Robust error classification

src/rules/add-items.js

@@ -16,7 +16,7 @@ import { analyzeBoardItem } from './helpers/board-items-evaluator.js';
  */
 const VERIFY_DELAY_MS = 5000; // 5 second delay for eventual consistency
 
-async function processAddItems({ org, repos, monitoredUser, projectId, windowHours, seedItems = [], allowedOrgs = [] }, overrides = {}) {
+async function processAddItems({ org, repos, monitoredUser, projectId, windowHours, seedItems = [], allowedOrgs = [], since = undefined }, overrides = {}) {
   const {
     getRecentItemsFn = getRecentItems,
     processBoardItemRulesFn = processBoardItemRules,
@@ -30,7 +30,10 @@ async function processAddItems({ org, repos, monitoredUser, projectId, windowHou
   logger.info(`Starting item processing for user ${monitoredUser}`);
 
   // Get recent items from API
-  const apiItems = await getRecentItemsFn(org, repos, monitoredUser, windowHours, { allowedOrgs });
+  const apiItems = await getRecentItemsFn(org, repos, monitoredUser, windowHours, { 
+    allowedOrgs,
+    since 
+  });
 
   // Combine seed items from event with API results
   const items = [...seedItems, ...apiItems];

src/utils/watermark.js

@@ -0,0 +1,114 @@
+import * as cache from '@actions/cache';
+import * as core from '@actions/core';
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import os from 'node:os';
+import { log } from './log.js';
+
+const CACHE_KEY_PREFIX = 'project-sync-watermark-';
+const WATERMARK_FILE = 'last_sync_timestamp.txt';
+
+/**
+ * Get the last successful sync timestamp from GitHub Actions cache.
+ * Uses the project ID to scope the cache.
+ * 
+ * @param {string} projectId - The GitHub Project V2 ID
+ * @returns {Promise<string|null>} ISO timestamp or null if not found
+ */
+export async function getWatermark(projectId, overrides = {}) {
+  const {
+    cache: cacheModule = cache,
+    fs: fsModule = fs,
+    log: logger = log
+  } = overrides;
+
+  if (!projectId) {
+    logger.warning('No project ID provided for watermark retrieval');
+    return null;
+  }
+
+  // Use a sanitised project ID for the cache key
+  const safeProjectId = projectId.replace(/[^a-zA-Z0-9]/g, '_');
+  const primaryKey = `${CACHE_KEY_PREFIX}${safeProjectId}-`; // We use restore keys to find latest
+
+  const tmpDir = await fsModule.mkdtemp(path.join(os.tmpdir(), 'watermark-'));
+  const filePath = path.join(tmpDir, WATERMARK_FILE);
+
+  try {
+    logger.info(`Attempting to restore watermark for project ${projectId}...`);
+    // restoreCache will find the most recent cache entry starting with this prefix
+    const restoredKey = await cacheModule.restoreCache([tmpDir], `${primaryKey}${Date.now()}`, [primaryKey]);
+
+    if (restoredKey) {
+      const content = await fsModule.readFile(filePath, 'utf8');
+      const timestamp = content.trim();
+      logger.info(`Restored watermark: ${timestamp} (from key: ${restoredKey})`);
+
+      // Basic validation: ensure it's a valid date
+      if (!isNaN(Date.parse(timestamp))) {
+        return timestamp;
+      }
+      logger.warning(`Restored watermark "${timestamp}" is not a valid ISO date`);
+    } else {
+      logger.info('No existing watermark found in cache');
+    }
+  } catch (error) {
+    logger.error(`Failed to restore watermark from cache: ${error.message}`);
+    if (process.env.DEBUG) logger.debug(error.stack);
+  } finally {
+    try {
+      await fsModule.rm(tmpDir, { recursive: true, force: true });
+    } catch (e) {
+      // Ignore cleanup errors
+    }
+  }
+  return null;
+}
+
+/**
+ * Save the current sync timestamp to GitHub Actions cache.
+ * 
+ * @param {string} projectId - The GitHub Project V2 ID
+ * @param {string} timestamp - ISO timestamp to save
+ * @returns {Promise<void>}
+ */
+export async function saveWatermark(projectId, timestamp, overrides = {}) {
+  const {
+    cache: cacheModule = cache,
+    fs: fsModule = fs,
+    log: logger = log
+  } = overrides;
+
+  if (!projectId || !timestamp) {
+    logger.warning('Project ID and timestamp are required to save watermark');
+    return;
+  }
+
+  const safeProjectId = projectId.replace(/[^a-zA-Z0-9]/g, '_');
+  const cacheKey = `${CACHE_KEY_PREFIX}${safeProjectId}-${Date.now()}`;
+
+  const tmpDir = await fsModule.mkdtemp(path.join(os.tmpdir(), 'watermark-'));
+  const filePath = path.join(tmpDir, WATERMARK_FILE);
+
+  try {
+    await fsModule.writeFile(filePath, timestamp);
+    logger.info(`Saving watermark ${timestamp} to cache with key ${cacheKey}...`);
+
+    // We don't want to fail the whole run if cache saving fails
+    await cacheModule.saveCache([tmpDir], cacheKey);
+    logger.info('Watermark saved successfully');
+  } catch (error) {
+    // Special handling for "Cache already exists" which shouldn't happen with our timestamping but just in case
+    if (error.name === 'ValidationError' && error.message.includes('already exists')) {
+      logger.info('Watermark cache already exists for this key, skipping.');
+    } else {
+      logger.error(`Failed to save watermark to cache: ${error.message}`);
+    }
+  } finally {
+    try {
+      await fsModule.rm(tmpDir, { recursive: true, force: true });
+    } catch (e) {
+      // Ignore cleanup errors
+    }
+  }
+}

test/github/get-recent-items-rate-limit.test.mjs

@@ -77,10 +77,10 @@ test('getRecentItems constructs assignee search query correctly', async () => {
   const assigneeCall = calls.find(call => call.includes('assignee:'));
   assert.ok(assigneeCall, 'Assignee search should be called');
   assert.ok(assigneeCall.includes('assignee:testuser'), 'Query should include assignee username');
-  assert.ok(assigneeCall.includes('updated:>'), 'Query should include updated timestamp filter');
+  assert.ok(assigneeCall.includes('updated:>='), 'Query should include updated timestamp filter');
 
   // Verify timestamp format (ISO 8601)
-  const timestampMatch = assigneeCall.match(/updated:>([^\s]+)/);
+  const timestampMatch = assigneeCall.match(/updated:>=([^\s]+)/);
   assert.ok(timestampMatch, 'Query should include timestamp in ISO format');
   assert.ok(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}/.test(timestampMatch[1]), 'Timestamp should be ISO 8601 format');
 });

test/github/since-parameter.test.mjs

@@ -0,0 +1,57 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { getRecentItems } from '../../src/github/api.js';
+
+test('getRecentItems - Gapless Sync Search Query Construction', async (t) => {
+  const calls = [];
+  const overrides = {
+    shouldProceedFn: async () => ({ proceed: true }),
+    withBackoffFn: async operation => operation(),
+    graphqlClient: async (_query, variables) => {
+      calls.push(variables.searchQuery);
+      return { search: { nodes: [] } };
+    }
+  };
+
+  await t.test('uses absolute since parameter when provided (Watermark Pattern)', async () => {
+    calls.length = 0;
+    const since = '2025-06-12T10:00:00Z';
+    await getRecentItems('org', ['repo1'], 'user', undefined, { overrides, since });
+
+    assert.ok(calls.length > 0, 'Search calls should be made');
+    for (const query of calls) {
+      assert.ok(query.includes(`updated:>=${since}`), `Query should include updated:>=${since}`);
+    }
+  });
+
+  await t.test('falls back to 24h window if no since provided', async () => {
+    calls.length = 0;
+    const now = Date.now();
+    await getRecentItems('org', ['repo1'], 'user', 24, { overrides });
+
+    assert.ok(calls.length > 0, 'Search calls should be made');
+    for (const query of calls) {
+      assert.ok(query.includes('updated:>='), `Query should include updated condition: ${query}`);
+      // Find timestamp in string and verify it's close to 24 hours ago
+      const timestampMatch = query.match(/updated:>=([^ ]+)/);
+      if (timestampMatch) {
+         const tDate = new Date(timestampMatch[1]);
+         const diff = now - tDate.getTime();
+         // Close enough to 24 hours (86.4M ms)
+         assert.ok(Math.abs(diff - 24 * 60 * 60 * 1000) < 60000, `Expected ~24h diff, got ${diff}`);
+      }
+    }
+  });
+
+  await t.test('is prioritized appropriately if since is present', async () => {
+    const watermarkSince = '2025-01-01T00:00:00Z';
+    // Repos and org-level searches should all use the watermark
+    await getRecentItems('bcgov', ['some-repo'], 'derek', 1, { overrides, since: watermarkSince });
+
+    const repoCall = calls.find(c => c.includes('repo:bcgov/some-repo'));
+    const authorCall = calls.find(c => c.includes('author:derek'));
+
+    assert.ok(repoCall.includes(`updated:>=${watermarkSince}`), 'Repo search should use watermark');
+    assert.ok(authorCall.includes(`updated:>=${watermarkSince}`), 'Author search should use watermark');
+  });
+});

test/utils/watermark.test.mjs

@@ -0,0 +1,79 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { getWatermark, saveWatermark } from '../../src/utils/watermark.js';
+
+test('Watermark Utility - GHA Cache Integration', async (t) => {
+  let mockCacheData = {};
+
+  const mockCache = {
+    restoreCache: async (paths, primaryKey, restoreKeys) => {
+      const prefix = restoreKeys[0];
+      if (mockCacheData[prefix]) {
+        return `${prefix}restored-from-mock`;
+      }
+      return null;
+    },
+    saveCache: async (paths, key) => {
+      // Extract prefix before the last dash
+      const prefix = key.substring(0, key.lastIndexOf('-') + 1);
+      mockCacheData[prefix] = 'mock-saved-data';
+      return 42;
+    }
+  };
+
+  const mockFs = {
+    mkdtemp: async (prefix) => `/tmp/mock-dir-${prefix.replace(/[^a-zA-Z0-9]/g, '_')}`,
+    writeFile: async () => {},
+    readFile: async () => '2024-01-01T00:00:00.000Z',
+    rm: async () => {}
+  };
+
+  const mockLog = {
+    info: () => {},
+    warning: () => {},
+    error: () => {},
+    debug: () => {}
+  };
+
+  await t.test('should return null when no watermark exists for project', async () => {
+    mockCacheData = {};
+    const watermark = await getWatermark('my-project', { cache: mockCache, fs: mockFs, log: mockLog });
+    assert.strictEqual(watermark, null);
+  });
+
+  await t.test('should save and retrieve watermark via cache prefix', async () => {
+    mockCacheData = {};
+    const timestamp = '2024-01-01T12:00:00.000Z';
+    await saveWatermark('my-project', timestamp, { cache: mockCache, fs: mockFs, log: mockLog });
+
+    // The utility should find the saved watermark even though the key has a timestamp
+    const watermark = await getWatermark('my-project', { cache: mockCache, fs: mockFs, log: mockLog });
+    assert.strictEqual(watermark, '2024-01-01T00:00:00.000Z');
+  });
+
+  await t.test('should handle project ID sanitization correctly', async () => {
+    mockCacheData = {};
+    const timestamp = '2024-01-02T12:00:00.000Z';
+    // Test with characters that must be sanitized for cache keys
+    const specialProjectId = 'org/projects:123';
+    await saveWatermark(specialProjectId, timestamp, { cache: mockCache, fs: mockFs, log: mockLog });
+
+    const watermark = await getWatermark(specialProjectId, { cache: mockCache, fs: mockFs, log: mockLog });
+    assert.strictEqual(watermark, '2024-01-01T00:00:00.000Z');
+
+    // Verify prefix was sanitized (org_projects_123)
+    const prefixes = Object.keys(mockCacheData);
+    assert.ok(prefixes[0].includes('org_projects_123'), `Prefix ${prefixes[0]} should be sanitized`);
+  });
+
+  await t.test('should return null if restored timestamp is invalid', async () => {
+    const corruptedFs = {
+      ...mockFs,
+      readFile: async () => 'not-a-valid-timestamp'
+    };
+    mockCacheData['project-sync-watermark-bad_data-'] = 'saved';
+
+    const watermark = await getWatermark('bad-data', { cache: mockCache, fs: corruptedFs, log: mockLog });
+    assert.strictEqual(watermark, null);
+  });
+});