Skip to content

feat: add daily import zetkin job #404

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
joaquimds merged 1 commit into from
Apr 1, 2026
+28 −0
Merged

feat: add daily import zetkin job #404

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/instrumentation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,8 @@ export async function register() {
const { schedule } = await import("@/server/services/queue");
// Run refreshWebhooks at 3:00am every day (see https://timgit.github.io/pg-boss/#/./api/scheduling)
await schedule("0 3 * * *", "refreshWebhooks", {});
// Re-import all Zetkin data sources at 4:00am every day
await schedule("0 4 * * *", "importZetkinDataSources", {});
Comment on lines 38 to +40
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

schedule() ultimately calls boss.schedule(queue, ...). With pg-boss, the first argument is the schedule/job name and must be unique; scheduling twice with the same name (here: the default queue name) will overwrite/update the existing schedule. As written, the 4:00am schedule will replace the 3:00am refreshWebhooks schedule (or vice-versa), so only one of them will actually run. Use distinct schedule names per task (e.g., incorporate task into the scheduled job name), or change the scheduling approach so multiple cron tasks can coexist on the same queue.

Suggested change
await schedule("0 3 * * *", "refreshWebhooks", {});
// Re-import all Zetkin data sources at 4:00am every day
await schedule("0 4 * * *", "importZetkinDataSources", {});
await schedule("0 3 * * *", "refreshWebhooks", {}, "refreshWebhooksSchedule");
// Re-import all Zetkin data sources at 4:00am every day
await schedule("0 4 * * *", "importZetkinDataSources", {}, "importZetkinDataSourcesSchedule");

Copilot uses AI. Check for mistakes.

logger.info("Started");
}
Expand Down
24 changes: 24 additions & 0 deletions src/server/jobs/importZetkinDataSources.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
import { DataSourceType } from "@/models/DataSource";
import { findDataSourcesByType } from "@/server/repositories/DataSource";
import logger from "../services/logger";
import { enqueue } from "../services/queue";

const importZetkinDataSources = async (): Promise<boolean> => {
const zetkinDataSources = await findDataSourcesByType(DataSourceType.Zetkin);
for (const source of zetkinDataSources) {
try {
await enqueue("importDataSource", source.id, {
dataSourceId: source.id,
});
Comment on lines +6 to +12
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

New scheduled job logic is introduced here, but there’s no test coverage to verify it enqueues importDataSource for each Zetkin data source (and handles failures as intended). Given there are existing tests for other jobs under tests/** (e.g. importDataSource), please add a unit/feature test that seeds Zetkin data sources and asserts the correct enqueue() calls / scheduled behavior.

Copilot uses AI. Check for mistakes.
} catch (error) {
logger.warn(
`Failed to enqueue import for Zetkin data source ${source.id}`,
{ error },
);
continue;
}
}
return true;
Comment on lines +8 to +21
Copy link

Copilot AI Apr 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This job swallows any enqueue() error and still returns true, which means the scheduled job will be marked successful even if no imports were enqueued (e.g., if PgBoss is temporarily unavailable). Consider letting unexpected enqueue() failures fail the job (so pg-boss retries), or at least tracking whether any enqueue failed and returning false/throwing at the end to avoid silently skipping a day's imports.

Copilot uses AI. Check for mistakes.
};

export default importZetkinDataSources;
2 changes: 2 additions & 0 deletions src/server/services/worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import enrichDataRecords from "@/server/jobs/enrichDataRecords";
import enrichDataSource from "@/server/jobs/enrichDataSource";
import importDataRecords from "@/server/jobs/importDataRecords";
import importDataSource from "@/server/jobs/importDataSource";
import importZetkinDataSources from "@/server/jobs/importZetkinDataSources";
import refreshWebhooks from "@/server/jobs/refreshWebhooks";
import removeEnrichmentColumns from "@/server/jobs/removeEnrichmentColumns";
import tagDataSource from "@/server/jobs/tagDataSource";
Expand All @@ -14,6 +15,7 @@ const taskHandlers: Record<string, (args: object | null) => Promise<boolean>> =
enrichDataRecords,
importDataSource,
importDataRecords,
importZetkinDataSources,
refreshWebhooks,
removeEnrichmentColumns,
tagDataSource,
Expand Down
Loading