This commit is contained in:
Ben Senescu
GitHub
parent
e2c84803f2
commit
f048dc3bd1
@ -10,7 +10,7 @@ data, or sensitive paths.
|
||||
|
||||
## Open
|
||||
|
||||
- [ ] `2026-08-05T20:59:09Z` — `codex` — The documented `pnpm seed:rank-tracking` command fails before opening local D1 because `scripts/seed-rank-tracking.ts` imports the provider-aware `src/db/schema` barrel and plain `tsx` cannot load the resulting `cloudflare:workers` URL. Keep the seed script on dialect-local schema imports or run it through a Workers-compatible execution path.
|
||||
- [ ] `2026-08-05T20:59:09Z` — `codex` — The documented `pnpm seed:rank-tracking` command fails before opening local D1 because `scripts/seed-rank-tracking.ts` imports the provider-aware `src/db/schema` barrel and plain `tsx` cannot load the resulting `cloudflare:workers` URL. Keep the seed script on dialect-local schema imports or run it through a Workers-compatible execution path. (Workaround: seed via raw SQL with `wrangler d1 execute DB --local`.)
|
||||
- [ ] `2026-08-01T16:28:36Z` — `claude` — web's pinned wrangler 4.71.0 fails `kv namespace create` with a bare "Authentication error [code: 10000]" even though the OAuth token has workers_kv write scope; wrangler@4.118.0 succeeds with identical auth. Fix: bump wrangler in web/package.json.
|
||||
- [ ] `2026-07-20T20:08:28Z` — `claude` — In a fresh git worktree, `oxlint --type-aware` crashes with `Cannot find module '@oxlint/binding-darwin-arm64'` — the platform-specific optional dep is missing from the worktree's node_modules while tsc/prettier work fine, and plain `pnpm install` reports up-to-date without restoring it; `pnpm install --force` (~22s) fixes it. Worth making the worktree-setup hook (or a documented step) run the forced install so lint doesn't die on fresh worktrees.
|
||||
- [ ] `2026-07-19T04:06:52Z` — `codex` — `pnpm --dir web build` fails with `vite: command not found` when `web/node_modules` is absent, despite the root toolchain being installed. Document or enforce the package-local install required before validating the `web/` subpackage.
|
||||
|
||||
@ -234,6 +234,12 @@ function DomainRow({
|
||||
Scheduled check skipped — insufficient credits
|
||||
</p>
|
||||
)}
|
||||
{summary.lastSkipReason === "plan_required" && (
|
||||
<p className="flex items-center gap-1 text-xs text-warning">
|
||||
<AlertTriangle className="size-3" />
|
||||
Scheduled check skipped — paid plan required
|
||||
</p>
|
||||
)}
|
||||
</div>
|
||||
<div className="hidden sm:flex items-center gap-6 text-sm pointer-events-none">
|
||||
{summary.keywordCount > 0 && (
|
||||
|
||||
@ -11,7 +11,7 @@ type BatchStatement = Parameters<typeof d1Db.batch>[0][number];
|
||||
// D1 caps bound parameters at ~100 per statement; keep batches bounded so each
|
||||
// runBatch call stays under that limit. (Postgres allows far more, but the same
|
||||
// chunk size is harmless there.)
|
||||
const DB_BATCH_SIZE = 100;
|
||||
export const DB_BATCH_SIZE = 100;
|
||||
|
||||
/**
|
||||
* Run a set of write statements atomically on either backend.
|
||||
|
||||
@ -189,7 +189,12 @@ export default {
|
||||
_ctx: ExecutionContext,
|
||||
) {
|
||||
// Scope a per-request Postgres client for the cron run (no-op in D1 mode).
|
||||
await withPgClient(() => runScheduledRankChecks(env));
|
||||
// Caught so a rank-tracking failure can't suppress the audit watchdog below.
|
||||
try {
|
||||
await withPgClient(() => runScheduledRankChecks(env));
|
||||
} catch (err) {
|
||||
console.error("[cron] Scheduled rank checks failed:", err);
|
||||
}
|
||||
// Watchdog: reconcile audits stuck in "running" whose workflow died
|
||||
// without reaching mark-failed (OOM/CPU kills, expired instances).
|
||||
await withPgClient(() => reconcileStaleAudits());
|
||||
|
||||
@ -63,10 +63,24 @@ describe("subscription billing", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("returns false when org lacks paid plan", async () => {
|
||||
it("returns false without retrying when org lacks paid plan", async () => {
|
||||
checkMock.mockResolvedValue({ allowed: false });
|
||||
|
||||
await expect(customerHasPaidPlan("org_123")).resolves.toBe(false);
|
||||
expect(checkMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("recovers from a degraded negative read when retryDenied is set", async () => {
|
||||
vi.useFakeTimers();
|
||||
checkMock
|
||||
.mockResolvedValueOnce({ allowed: false })
|
||||
.mockResolvedValueOnce({ allowed: true });
|
||||
|
||||
const result = customerHasPaidPlan("org_123", { retryDenied: true });
|
||||
await vi.runAllTimersAsync();
|
||||
|
||||
await expect(result).resolves.toBe(true);
|
||||
expect(checkMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("retries a missing monthly balance once", async () => {
|
||||
|
||||
@ -63,13 +63,27 @@ export async function getOrCreateOrganizationCustomer(
|
||||
return { id: customer.id };
|
||||
}
|
||||
|
||||
export async function customerHasPaidPlan(customerId: string) {
|
||||
export async function customerHasPaidPlan(
|
||||
customerId: string,
|
||||
opts: { retryDenied?: boolean } = {},
|
||||
) {
|
||||
const result = await autumn.check({
|
||||
customerId,
|
||||
featureId: AUTUMN_PAID_PLAN_FEATURE_ID,
|
||||
});
|
||||
if (result.allowed || !opts.retryDenied) return result.allowed;
|
||||
|
||||
return result.allowed;
|
||||
// Autumn sometimes returns degraded entitlement data in a successful
|
||||
// response (see the balance retry in getUsageCreditsRemaining). Where a
|
||||
// false negative does lasting damage — the scheduler would advance a paying
|
||||
// org's schedule and flag "plan_required" — callers opt into one re-check.
|
||||
// Interactive deny paths skip it to stay fast for genuinely free users.
|
||||
await new Promise((resolve) => setTimeout(resolve, 300));
|
||||
const retry = await autumn.check({
|
||||
customerId,
|
||||
featureId: AUTUMN_PAID_PLAN_FEATURE_ID,
|
||||
});
|
||||
return retry.allowed;
|
||||
}
|
||||
|
||||
export async function customerHasManagedAccess(customerId: string) {
|
||||
|
||||
@ -0,0 +1,278 @@
|
||||
import { createClient, type Client } from "@libsql/client";
|
||||
import { drizzle } from "drizzle-orm/libsql";
|
||||
import { eq } from "drizzle-orm";
|
||||
import {
|
||||
afterAll,
|
||||
beforeAll,
|
||||
beforeEach,
|
||||
describe,
|
||||
expect,
|
||||
it,
|
||||
vi,
|
||||
} from "vitest";
|
||||
import { rankTrackingConfigs } from "@/db/schema";
|
||||
import type * as RankTrackingRepositoryModule from "./RankTrackingRepository";
|
||||
|
||||
// Real in-memory SQLite so the due-query ordering, the manual-interval filter,
|
||||
// and claimDueConfig's compare-and-set run against actual SQL — the parts the
|
||||
// mocked service tests can't see.
|
||||
|
||||
vi.mock("cloudflare:workers", () => ({
|
||||
env: { DATABASE_PROVIDER: "d1" },
|
||||
}));
|
||||
|
||||
let client: Client;
|
||||
let testDb: ReturnType<typeof drizzle>;
|
||||
let RankTrackingRepository: typeof RankTrackingRepositoryModule.RankTrackingRepository;
|
||||
|
||||
beforeAll(async () => {
|
||||
client = createClient({ url: "file::memory:" });
|
||||
testDb = drizzle(client);
|
||||
vi.doMock("@/db", () => ({ db: testDb }));
|
||||
|
||||
await client.executeMultiple(`
|
||||
CREATE TABLE projects (
|
||||
id TEXT PRIMARY KEY,
|
||||
organization_id TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
domain TEXT,
|
||||
location_code INTEGER NOT NULL DEFAULT 2840,
|
||||
language_code TEXT NOT NULL DEFAULT 'en',
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
archived_at TEXT
|
||||
);
|
||||
CREATE TABLE rank_tracking_configs (
|
||||
id TEXT PRIMARY KEY,
|
||||
project_id TEXT NOT NULL,
|
||||
domain TEXT NOT NULL,
|
||||
location_code INTEGER NOT NULL DEFAULT 2840,
|
||||
language_code TEXT NOT NULL DEFAULT 'en',
|
||||
devices TEXT NOT NULL DEFAULT 'both',
|
||||
serp_depth INTEGER NOT NULL,
|
||||
schedule_interval TEXT NOT NULL DEFAULT 'weekly',
|
||||
location_name TEXT,
|
||||
is_active INTEGER NOT NULL DEFAULT 1,
|
||||
last_checked_at TEXT,
|
||||
next_check_at TEXT,
|
||||
last_skip_reason TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE TABLE rank_tracking_keywords (
|
||||
id TEXT PRIMARY KEY,
|
||||
config_id TEXT NOT NULL,
|
||||
keyword TEXT NOT NULL,
|
||||
search_volume INTEGER,
|
||||
keyword_difficulty INTEGER,
|
||||
cpc REAL,
|
||||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
`);
|
||||
|
||||
({ RankTrackingRepository } = await import("./RankTrackingRepository"));
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
client.close();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await client.executeMultiple(`
|
||||
DELETE FROM rank_tracking_keywords;
|
||||
DELETE FROM rank_tracking_configs;
|
||||
DELETE FROM projects;
|
||||
`);
|
||||
});
|
||||
|
||||
const NOW = "2026-08-01T12:00:00.000Z";
|
||||
|
||||
async function seedProject(id: string, archivedAt: string | null = null) {
|
||||
await client.execute({
|
||||
sql: "INSERT INTO projects (id, organization_id, name, archived_at) VALUES (?, ?, ?, ?)",
|
||||
args: [id, `org_${id}`, id, archivedAt],
|
||||
});
|
||||
}
|
||||
|
||||
async function seedConfig(input: {
|
||||
id: string;
|
||||
projectId?: string;
|
||||
nextCheckAt?: string | null;
|
||||
scheduleInterval?: string;
|
||||
isActive?: number;
|
||||
lastSkipReason?: string | null;
|
||||
}) {
|
||||
await client.execute({
|
||||
sql: `INSERT INTO rank_tracking_configs
|
||||
(id, project_id, domain, serp_depth, schedule_interval, is_active, next_check_at, last_skip_reason)
|
||||
VALUES (?, ?, ?, 20, ?, ?, ?, ?)`,
|
||||
args: [
|
||||
input.id,
|
||||
input.projectId ?? "proj_1",
|
||||
`${input.id}.example.com`,
|
||||
input.scheduleInterval ?? "daily",
|
||||
input.isActive ?? 1,
|
||||
input.nextCheckAt === undefined
|
||||
? "2026-07-01T05:00:00.000Z"
|
||||
: input.nextCheckAt,
|
||||
input.lastSkipReason ?? null,
|
||||
],
|
||||
});
|
||||
}
|
||||
|
||||
async function getConfigRow(id: string) {
|
||||
const rows = await testDb
|
||||
.select({
|
||||
nextCheckAt: rankTrackingConfigs.nextCheckAt,
|
||||
lastSkipReason: rankTrackingConfigs.lastSkipReason,
|
||||
})
|
||||
.from(rankTrackingConfigs)
|
||||
.where(eq(rankTrackingConfigs.id, id));
|
||||
return rows[0];
|
||||
}
|
||||
|
||||
describe("getDueConfigsWithOrganization", () => {
|
||||
it("returns due rows oldest-first and excludes manual, inactive, future, and archived rows", async () => {
|
||||
await seedProject("proj_1");
|
||||
await seedProject("proj_archived", "2026-07-15T00:00:00.000Z");
|
||||
await seedConfig({
|
||||
id: "cfg_newer",
|
||||
nextCheckAt: "2026-07-02T05:00:00.000Z",
|
||||
});
|
||||
await seedConfig({
|
||||
id: "cfg_oldest",
|
||||
nextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
});
|
||||
// Same timestamp as cfg_oldest — id breaks the tie deterministically.
|
||||
await seedConfig({
|
||||
id: "cfg_tie",
|
||||
nextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
});
|
||||
await seedConfig({ id: "cfg_manual", scheduleInterval: "manual" });
|
||||
await seedConfig({ id: "cfg_inactive", isActive: 0 });
|
||||
await seedConfig({
|
||||
id: "cfg_future",
|
||||
nextCheckAt: "2027-01-01T05:00:00.000Z",
|
||||
});
|
||||
await seedConfig({ id: "cfg_null_due", nextCheckAt: null });
|
||||
await seedConfig({ id: "cfg_archived", projectId: "proj_archived" });
|
||||
|
||||
const due = await RankTrackingRepository.getDueConfigsWithOrganization(NOW);
|
||||
|
||||
expect(due.map((c) => c.id)).toEqual([
|
||||
"cfg_oldest",
|
||||
"cfg_tie",
|
||||
"cfg_newer",
|
||||
]);
|
||||
expect(due[0].organizationId).toBe("org_proj_1");
|
||||
});
|
||||
});
|
||||
|
||||
describe("claimDueConfig", () => {
|
||||
it("advances the schedule and writes the skip reason when the token matches", async () => {
|
||||
await seedProject("proj_1");
|
||||
await seedConfig({ id: "cfg_1", nextCheckAt: "2026-07-01T05:00:00.000Z" });
|
||||
|
||||
const claimed = await RankTrackingRepository.claimDueConfig({
|
||||
configId: "cfg_1",
|
||||
projectId: "proj_1",
|
||||
observedNextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
nextCheckAt: "2026-08-02T05:00:00.000Z",
|
||||
lastSkipReason: "plan_required",
|
||||
});
|
||||
|
||||
expect(claimed).toBe(true);
|
||||
expect(await getConfigRow("cfg_1")).toEqual({
|
||||
nextCheckAt: "2026-08-02T05:00:00.000Z",
|
||||
lastSkipReason: "plan_required",
|
||||
});
|
||||
});
|
||||
|
||||
it("refuses when the observed token no longer matches, leaving the row untouched", async () => {
|
||||
await seedProject("proj_1");
|
||||
await seedConfig({
|
||||
id: "cfg_1",
|
||||
nextCheckAt: "2026-07-20T05:00:00.000Z",
|
||||
lastSkipReason: "insufficient_credits",
|
||||
});
|
||||
|
||||
const claimed = await RankTrackingRepository.claimDueConfig({
|
||||
configId: "cfg_1",
|
||||
projectId: "proj_1",
|
||||
observedNextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
nextCheckAt: "2026-08-02T05:00:00.000Z",
|
||||
lastSkipReason: null,
|
||||
});
|
||||
|
||||
expect(claimed).toBe(false);
|
||||
expect(await getConfigRow("cfg_1")).toEqual({
|
||||
nextCheckAt: "2026-07-20T05:00:00.000Z",
|
||||
lastSkipReason: "insufficient_credits",
|
||||
});
|
||||
});
|
||||
|
||||
it("refuses deactivated rows and preserves the skip reason when omitted", async () => {
|
||||
await seedProject("proj_1");
|
||||
await seedConfig({ id: "cfg_off", isActive: 0 });
|
||||
await seedConfig({
|
||||
id: "cfg_restore",
|
||||
nextCheckAt: "2026-08-02T05:00:00.000Z",
|
||||
lastSkipReason: "insufficient_credits",
|
||||
});
|
||||
|
||||
expect(
|
||||
await RankTrackingRepository.claimDueConfig({
|
||||
configId: "cfg_off",
|
||||
projectId: "proj_1",
|
||||
observedNextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
nextCheckAt: "2026-08-02T05:00:00.000Z",
|
||||
}),
|
||||
).toBe(false);
|
||||
|
||||
// Restore-shaped claim: no lastSkipReason key — the existing value stays.
|
||||
expect(
|
||||
await RankTrackingRepository.claimDueConfig({
|
||||
configId: "cfg_restore",
|
||||
projectId: "proj_1",
|
||||
observedNextCheckAt: "2026-08-02T05:00:00.000Z",
|
||||
nextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(await getConfigRow("cfg_restore")).toEqual({
|
||||
nextCheckAt: "2026-07-01T05:00:00.000Z",
|
||||
lastSkipReason: "insufficient_credits",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("getKeywordCountsForConfigs", () => {
|
||||
it("groups counts by config and omits configs with no keywords", async () => {
|
||||
await seedProject("proj_1");
|
||||
await seedConfig({ id: "cfg_a" });
|
||||
await seedConfig({ id: "cfg_b" });
|
||||
for (const [id, configId] of [
|
||||
["kw1", "cfg_a"],
|
||||
["kw2", "cfg_a"],
|
||||
["kw3", "cfg_b"],
|
||||
] as const) {
|
||||
await client.execute({
|
||||
sql: "INSERT INTO rank_tracking_keywords (id, config_id, keyword) VALUES (?, ?, ?)",
|
||||
args: [id, configId, id],
|
||||
});
|
||||
}
|
||||
|
||||
const counts = await RankTrackingRepository.getKeywordCountsForConfigs([
|
||||
"cfg_a",
|
||||
"cfg_b",
|
||||
"cfg_missing",
|
||||
]);
|
||||
|
||||
expect(counts).toEqual(
|
||||
new Map([
|
||||
["cfg_a", 2],
|
||||
["cfg_b", 1],
|
||||
]),
|
||||
);
|
||||
expect(await RankTrackingRepository.getKeywordCountsForConfigs([])).toEqual(
|
||||
new Map(),
|
||||
);
|
||||
});
|
||||
});
|
||||
@ -1,4 +1,15 @@
|
||||
import { and, count, desc, eq, inArray, isNull, lte, max } from "drizzle-orm";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
count,
|
||||
desc,
|
||||
eq,
|
||||
inArray,
|
||||
isNull,
|
||||
lte,
|
||||
max,
|
||||
ne,
|
||||
} from "drizzle-orm";
|
||||
import type { InferInsertModel } from "drizzle-orm";
|
||||
import { db } from "@/db";
|
||||
import {
|
||||
@ -8,7 +19,8 @@ import {
|
||||
rankTrackingKeywords,
|
||||
projects,
|
||||
} from "@/db/schema";
|
||||
import { executeInBatches } from "@/db/runBatch";
|
||||
import { DB_BATCH_SIZE, executeInBatches } from "@/db/runBatch";
|
||||
import type { RankTrackingSkipReason } from "@/shared/rank-tracking";
|
||||
import {
|
||||
getLatestSnapshotsForKeywords,
|
||||
getSnapshotsBeforeDate,
|
||||
@ -104,30 +116,81 @@ async function updateConfig(
|
||||
}
|
||||
|
||||
async function getDueConfigsWithOrganization(nowIso: string) {
|
||||
return db
|
||||
.select({
|
||||
id: rankTrackingConfigs.id,
|
||||
projectId: rankTrackingConfigs.projectId,
|
||||
domain: rankTrackingConfigs.domain,
|
||||
locationCode: rankTrackingConfigs.locationCode,
|
||||
languageCode: rankTrackingConfigs.languageCode,
|
||||
locationName: rankTrackingConfigs.locationName,
|
||||
devices: rankTrackingConfigs.devices,
|
||||
serpDepth: rankTrackingConfigs.serpDepth,
|
||||
scheduleInterval: rankTrackingConfigs.scheduleInterval,
|
||||
nextCheckAt: rankTrackingConfigs.nextCheckAt,
|
||||
organizationId: projects.organizationId,
|
||||
return (
|
||||
db
|
||||
.select({
|
||||
id: rankTrackingConfigs.id,
|
||||
projectId: rankTrackingConfigs.projectId,
|
||||
domain: rankTrackingConfigs.domain,
|
||||
locationCode: rankTrackingConfigs.locationCode,
|
||||
languageCode: rankTrackingConfigs.languageCode,
|
||||
locationName: rankTrackingConfigs.locationName,
|
||||
devices: rankTrackingConfigs.devices,
|
||||
serpDepth: rankTrackingConfigs.serpDepth,
|
||||
scheduleInterval: rankTrackingConfigs.scheduleInterval,
|
||||
nextCheckAt: rankTrackingConfigs.nextCheckAt,
|
||||
organizationId: projects.organizationId,
|
||||
})
|
||||
.from(rankTrackingConfigs)
|
||||
.innerJoin(projects, eq(rankTrackingConfigs.projectId, projects.id))
|
||||
.where(
|
||||
and(
|
||||
eq(rankTrackingConfigs.isActive, true),
|
||||
// A manual config can keep a stale non-null next_check_at; without this
|
||||
// it would be selected every tick and never advanced.
|
||||
ne(rankTrackingConfigs.scheduleInterval, "manual"),
|
||||
lte(rankTrackingConfigs.nextCheckAt, nowIso),
|
||||
isNull(projects.archivedAt),
|
||||
),
|
||||
)
|
||||
// Oldest first so a large backlog drains in order instead of the same
|
||||
// arbitrary rows filling every batch. `lte` already excludes NULL, so both
|
||||
// ordering columns are non-null and SQLite/Postgres agree.
|
||||
.orderBy(
|
||||
asc(rankTrackingConfigs.nextCheckAt),
|
||||
asc(rankTrackingConfigs.id),
|
||||
)
|
||||
.limit(200)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Conditionally advance a due config's schedule, returning false when the
|
||||
* config changed underneath us (manual edit, deactivation).
|
||||
*
|
||||
* `next_check_at` equality is the compare-and-set token. `schedule_interval` is
|
||||
* deliberately absent from the predicate: every schedule edit rewrites
|
||||
* `next_check_at` (updateConfig recomputes it, or nulls it for "manual"), so
|
||||
* the timestamp check already detects interval changes.
|
||||
*
|
||||
* `lastSkipReason` is written only when the caller passes it — the restore
|
||||
* path omits it so it can't clobber a reason the blocking run just wrote.
|
||||
*/
|
||||
async function claimDueConfig(input: {
|
||||
configId: string;
|
||||
projectId: string;
|
||||
observedNextCheckAt: string;
|
||||
nextCheckAt: string;
|
||||
lastSkipReason?: RankTrackingSkipReason | null;
|
||||
}): Promise<boolean> {
|
||||
const claimed = await db
|
||||
.update(rankTrackingConfigs)
|
||||
.set({
|
||||
nextCheckAt: input.nextCheckAt,
|
||||
...(input.lastSkipReason !== undefined && {
|
||||
lastSkipReason: input.lastSkipReason,
|
||||
}),
|
||||
})
|
||||
.from(rankTrackingConfigs)
|
||||
.innerJoin(projects, eq(rankTrackingConfigs.projectId, projects.id))
|
||||
.where(
|
||||
and(
|
||||
eq(rankTrackingConfigs.id, input.configId),
|
||||
eq(rankTrackingConfigs.projectId, input.projectId),
|
||||
eq(rankTrackingConfigs.isActive, true),
|
||||
lte(rankTrackingConfigs.nextCheckAt, nowIso),
|
||||
isNull(projects.archivedAt),
|
||||
eq(rankTrackingConfigs.nextCheckAt, input.observedNextCheckAt),
|
||||
),
|
||||
)
|
||||
.limit(50);
|
||||
.returning({ id: rankTrackingConfigs.id });
|
||||
return claimed.length > 0;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@ -297,22 +360,7 @@ async function getConfigSummaries(projectId: string) {
|
||||
const configs = await getConfigsForProject(projectId);
|
||||
if (configs.length === 0) return [];
|
||||
|
||||
// Batch: keyword counts grouped by config
|
||||
const kwCounts = await db
|
||||
.select({
|
||||
configId: rankTrackingKeywords.configId,
|
||||
value: count(),
|
||||
})
|
||||
.from(rankTrackingKeywords)
|
||||
.where(
|
||||
inArray(
|
||||
rankTrackingKeywords.configId,
|
||||
configs.map((c) => c.id),
|
||||
),
|
||||
)
|
||||
.groupBy(rankTrackingKeywords.configId);
|
||||
|
||||
const kwCountMap = new Map(kwCounts.map((r) => [r.configId, r.value]));
|
||||
const kwCountMap = await getKeywordCountsForConfigs(configs.map((c) => c.id));
|
||||
|
||||
// Subquery: latest startedAt per config
|
||||
const latestStarted = db
|
||||
@ -395,6 +443,22 @@ async function getKeywordCountForConfig(configId: string) {
|
||||
return rows[0]?.value ?? 0;
|
||||
}
|
||||
|
||||
/** Keyword counts keyed by config id. Configs with no keywords are absent. */
|
||||
async function getKeywordCountsForConfigs(configIds: string[]) {
|
||||
// Chunked so the IN list stays under D1's ~100 bound-parameter cap.
|
||||
const counts = new Map<string, number>();
|
||||
for (let i = 0; i < configIds.length; i += DB_BATCH_SIZE) {
|
||||
const chunk = configIds.slice(i, i + DB_BATCH_SIZE);
|
||||
const rows = await db
|
||||
.select({ configId: rankTrackingKeywords.configId, value: count() })
|
||||
.from(rankTrackingKeywords)
|
||||
.where(inArray(rankTrackingKeywords.configId, chunk))
|
||||
.groupBy(rankTrackingKeywords.configId);
|
||||
for (const row of rows) counts.set(row.configId, row.value);
|
||||
}
|
||||
return counts;
|
||||
}
|
||||
|
||||
export const RankTrackingRepository = {
|
||||
getConfigsForProject,
|
||||
getConfigById,
|
||||
@ -402,6 +466,7 @@ export const RankTrackingRepository = {
|
||||
createConfig,
|
||||
updateConfig,
|
||||
getDueConfigsWithOrganization,
|
||||
claimDueConfig,
|
||||
tryCreateRun,
|
||||
updateRun,
|
||||
getRunById,
|
||||
@ -414,6 +479,7 @@ export const RankTrackingRepository = {
|
||||
removeKeywordsFromConfig,
|
||||
updateKeywordMetrics,
|
||||
getKeywordCountForConfig,
|
||||
getKeywordCountsForConfigs,
|
||||
getConfigSummaries,
|
||||
getLatestSnapshotsForKeywords,
|
||||
getSnapshotsBeforeDate,
|
||||
|
||||
@ -0,0 +1,354 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
type DueConfigRow = {
|
||||
id: string;
|
||||
projectId: string;
|
||||
domain: string;
|
||||
locationCode: number;
|
||||
languageCode: string;
|
||||
locationName: string | null;
|
||||
devices: "both" | "desktop" | "mobile";
|
||||
serpDepth: number;
|
||||
scheduleInterval: "daily" | "weekly" | "monthly" | "manual";
|
||||
nextCheckAt: string | null;
|
||||
organizationId: string;
|
||||
};
|
||||
|
||||
type ClaimInput = {
|
||||
configId: string;
|
||||
projectId: string;
|
||||
observedNextCheckAt: string;
|
||||
nextCheckAt: string;
|
||||
lastSkipReason?: string | null;
|
||||
};
|
||||
|
||||
type BeginResult =
|
||||
| { ok: true; runId: string }
|
||||
| { ok: false; reason: string; blockingRunId: string | null };
|
||||
|
||||
// Typed so assertions can read `mock.calls` without tripping the type-aware
|
||||
// lint rules on `any`.
|
||||
const mocks = vi.hoisted(() => ({
|
||||
getDueConfigsWithOrganization:
|
||||
vi.fn<(nowIso: string) => Promise<DueConfigRow[]>>(),
|
||||
getKeywordCountsForConfigs:
|
||||
vi.fn<(configIds: string[]) => Promise<Map<string, number>>>(),
|
||||
claimDueConfig: vi.fn<(input: ClaimInput) => Promise<boolean>>(),
|
||||
beginRankCheckRun:
|
||||
vi.fn<
|
||||
(input: {
|
||||
config: DueConfigRow;
|
||||
keywordsTotal: number;
|
||||
trigger: string;
|
||||
}) => Promise<BeginResult>
|
||||
>(),
|
||||
customerHasPaidPlan: vi.fn<(organizationId: string) => Promise<boolean>>(),
|
||||
isHostedServerAuthMode: vi.fn<() => Promise<boolean>>(),
|
||||
}));
|
||||
|
||||
vi.mock("cloudflare:workers", () => ({ env: {} }));
|
||||
vi.mock(
|
||||
"@/server/features/rank-tracking/repositories/RankTrackingRepository",
|
||||
() => ({
|
||||
RankTrackingRepository: {
|
||||
getDueConfigsWithOrganization: mocks.getDueConfigsWithOrganization,
|
||||
getKeywordCountsForConfigs: mocks.getKeywordCountsForConfigs,
|
||||
claimDueConfig: mocks.claimDueConfig,
|
||||
},
|
||||
}),
|
||||
);
|
||||
vi.mock("@/server/features/rank-tracking/services/rankCheckRunGuards", () => ({
|
||||
beginRankCheckRun: mocks.beginRankCheckRun,
|
||||
}));
|
||||
vi.mock("@/server/billing/subscription", () => ({
|
||||
customerHasPaidPlan: mocks.customerHasPaidPlan,
|
||||
}));
|
||||
vi.mock("@/server/lib/runtime-env", () => ({
|
||||
isHostedServerAuthMode: mocks.isHostedServerAuthMode,
|
||||
}));
|
||||
|
||||
// oxlint-disable-next-line typescript/no-unsafe-type-assertion -- test double for the workflow binding
|
||||
const testEnv = { RANK_CHECK_WORKFLOW: {} } as unknown as Env;
|
||||
|
||||
function dueConfig(overrides: Partial<DueConfigRow> = {}): DueConfigRow {
|
||||
return {
|
||||
id: "config_1",
|
||||
projectId: "project_1",
|
||||
domain: "acme.com",
|
||||
locationCode: 2840,
|
||||
languageCode: "en",
|
||||
locationName: null,
|
||||
devices: "both" as const,
|
||||
serpDepth: 20,
|
||||
scheduleInterval: "daily" as const,
|
||||
nextCheckAt: "2026-01-01T00:00:00.000Z",
|
||||
organizationId: "org_1",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
async function runTick() {
|
||||
const { runScheduledRankChecks } = await import("./scheduledRankChecks");
|
||||
await runScheduledRankChecks(testEnv);
|
||||
}
|
||||
|
||||
describe("runScheduledRankChecks", () => {
|
||||
beforeEach(() => {
|
||||
vi.resetModules();
|
||||
vi.resetAllMocks();
|
||||
mocks.isHostedServerAuthMode.mockResolvedValue(true);
|
||||
mocks.customerHasPaidPlan.mockResolvedValue(true);
|
||||
mocks.claimDueConfig.mockResolvedValue(true);
|
||||
mocks.beginRankCheckRun.mockResolvedValue({ ok: true, runId: "run_1" });
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([["config_1", 5]]),
|
||||
);
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([dueConfig()]);
|
||||
});
|
||||
|
||||
it("advances a free config with plan_required instead of starting a workflow", async () => {
|
||||
mocks.customerHasPaidPlan.mockResolvedValue(false);
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
configId: "config_1",
|
||||
projectId: "project_1",
|
||||
observedNextCheckAt: "2026-01-01T00:00:00.000Z",
|
||||
lastSkipReason: "plan_required",
|
||||
}),
|
||||
);
|
||||
const [claim] = mocks.claimDueConfig.mock.calls[0];
|
||||
expect(new Date(claim.nextCheckAt).getTime()).toBeGreaterThan(Date.now());
|
||||
expect(mocks.beginRankCheckRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("advances a zero-keyword config with no_keywords without consuming budget", async () => {
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([
|
||||
dueConfig({ id: "config_empty" }),
|
||||
dueConfig({ id: "config_2", nextCheckAt: "2026-01-02T00:00:00.000Z" }),
|
||||
]);
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([["config_2", 5]]),
|
||||
);
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
configId: "config_empty",
|
||||
lastSkipReason: "no_keywords",
|
||||
}),
|
||||
);
|
||||
// The empty config consumed no budget, so the paid one behind it still ran.
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ keywordsTotal: 5, trigger: "scheduled" }),
|
||||
);
|
||||
});
|
||||
|
||||
it("claims a paid config and starts its workflow", async () => {
|
||||
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
configId: "config_1",
|
||||
observedNextCheckAt: "2026-01-01T00:00:00.000Z",
|
||||
lastSkipReason: null,
|
||||
}),
|
||||
);
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
// 5 keywords × both devices = 10 task units.
|
||||
expect(logSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
event: "rank_tracking_scheduler_summary",
|
||||
candidates: 1,
|
||||
started: 1,
|
||||
unitsStarted: 10,
|
||||
stoppedByBudget: false,
|
||||
configErrors: 0,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("stops admitting configs once the task-unit budget is spent", async () => {
|
||||
// 400 keywords × 2 devices = 800 units — admitted anyway as the tick's
|
||||
// first start (oversized exemption), then exhausts the 200-unit budget.
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([
|
||||
dueConfig({ id: "config_big" }),
|
||||
dueConfig({ id: "config_next", nextCheckAt: "2026-01-02T00:00:00.000Z" }),
|
||||
]);
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([
|
||||
["config_big", 400],
|
||||
["config_next", 5],
|
||||
]),
|
||||
);
|
||||
|
||||
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.customerHasPaidPlan).toHaveBeenCalledTimes(1);
|
||||
expect(logSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ stoppedByBudget: true, unitsStarted: 800 }),
|
||||
);
|
||||
});
|
||||
|
||||
it("skips a config whose claim lost to a concurrent edit", async () => {
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([
|
||||
dueConfig({ id: "config_edited" }),
|
||||
dueConfig({ id: "config_2", nextCheckAt: "2026-01-02T00:00:00.000Z" }),
|
||||
]);
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([
|
||||
["config_edited", 5],
|
||||
["config_2", 5],
|
||||
]),
|
||||
);
|
||||
mocks.claimDueConfig.mockResolvedValueOnce(false);
|
||||
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.beginRankCheckRun.mock.calls[0][0].config.id).toBe("config_2");
|
||||
expect(logSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ concurrentChangeSkips: 1, started: 1 }),
|
||||
);
|
||||
});
|
||||
|
||||
it("contains a per-config failure and still processes the rest of the tick", async () => {
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([
|
||||
dueConfig({ id: "config_bad" }),
|
||||
dueConfig({ id: "config_ok", nextCheckAt: "2026-01-02T00:00:00.000Z" }),
|
||||
]);
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([
|
||||
["config_bad", 5],
|
||||
["config_ok", 5],
|
||||
]),
|
||||
);
|
||||
mocks.claimDueConfig.mockRejectedValueOnce(new Error("db blip"));
|
||||
// A tick with errors logs its summary at error level.
|
||||
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.beginRankCheckRun.mock.calls[0][0].config.id).toBe(
|
||||
"config_ok",
|
||||
);
|
||||
expect(errorSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configErrors: 1, started: 1 }),
|
||||
);
|
||||
});
|
||||
|
||||
it("defers a config whose units would overflow the remaining budget", async () => {
|
||||
// First config fits (10 units); the second would overflow (800), so it is
|
||||
// deferred rather than admitted just because budget remains.
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([
|
||||
dueConfig({ id: "config_small" }),
|
||||
dueConfig({ id: "config_big", nextCheckAt: "2026-01-02T00:00:00.000Z" }),
|
||||
]);
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([
|
||||
["config_small", 5],
|
||||
["config_big", 400],
|
||||
]),
|
||||
);
|
||||
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.beginRankCheckRun.mock.calls[0][0].config.id).toBe(
|
||||
"config_small",
|
||||
);
|
||||
expect(logSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ stoppedByBudget: true, unitsStarted: 10 }),
|
||||
);
|
||||
});
|
||||
|
||||
it("restores the original due time when a run is already active", async () => {
|
||||
mocks.beginRankCheckRun.mockResolvedValue({
|
||||
ok: false,
|
||||
reason: "already_running",
|
||||
blockingRunId: "run_blocking",
|
||||
});
|
||||
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledTimes(2);
|
||||
const [advance] = mocks.claimDueConfig.mock.calls[0];
|
||||
const [restore] = mocks.claimDueConfig.mock.calls[1];
|
||||
expect(restore).toEqual({
|
||||
configId: "config_1",
|
||||
projectId: "project_1",
|
||||
observedNextCheckAt: advance.nextCheckAt,
|
||||
nextCheckAt: "2026-01-01T00:00:00.000Z",
|
||||
});
|
||||
// Blocked configs leave no durable row state, so the summary names them.
|
||||
expect(logSpy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
alreadyRunning: 1,
|
||||
alreadyRunningConfigIds: ["config_1"],
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("leaves a config due when its plan check throws, and still processes other orgs", async () => {
|
||||
mocks.getDueConfigsWithOrganization.mockResolvedValue([
|
||||
dueConfig({ id: "config_a1", organizationId: "org_a" }),
|
||||
dueConfig({
|
||||
id: "config_a2",
|
||||
organizationId: "org_a",
|
||||
nextCheckAt: "2026-01-02T00:00:00.000Z",
|
||||
}),
|
||||
dueConfig({
|
||||
id: "config_b",
|
||||
organizationId: "org_b",
|
||||
nextCheckAt: "2026-01-03T00:00:00.000Z",
|
||||
}),
|
||||
]);
|
||||
mocks.getKeywordCountsForConfigs.mockResolvedValue(
|
||||
new Map([
|
||||
["config_a1", 5],
|
||||
["config_a2", 5],
|
||||
["config_b", 5],
|
||||
]),
|
||||
);
|
||||
mocks.customerHasPaidPlan.mockImplementation(async (orgId: string) => {
|
||||
if (orgId === "org_a") throw new Error("autumn down");
|
||||
return true;
|
||||
});
|
||||
vi.spyOn(console, "error").mockImplementation(() => {});
|
||||
|
||||
await runTick();
|
||||
|
||||
// One Autumn call per org per tick — org_a's rejection is memoized.
|
||||
expect(mocks.customerHasPaidPlan).toHaveBeenCalledTimes(2);
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.claimDueConfig).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configId: "config_b" }),
|
||||
);
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("makes no billing calls in self-hosted mode", async () => {
|
||||
mocks.isHostedServerAuthMode.mockResolvedValue(false);
|
||||
|
||||
await runTick();
|
||||
|
||||
expect(mocks.customerHasPaidPlan).not.toHaveBeenCalled();
|
||||
expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@ -4,90 +4,231 @@ import { customerHasPaidPlan } from "@/server/billing/subscription";
|
||||
import { isHostedServerAuthMode } from "@/server/lib/runtime-env";
|
||||
import {
|
||||
computeNextCheckAt,
|
||||
devicesCount,
|
||||
isScheduledRankTrackingInterval,
|
||||
} from "@/shared/rank-tracking";
|
||||
|
||||
// Work admitted per tick, in task units (keywords × devices). Admission
|
||||
// control, not a hard rate limit: the first start of a tick is always
|
||||
// admitted, so an oversized config can never starve. 200 units on the
|
||||
// 5-minute cron ≈ 57,600 units/day, ~9× current steady-state demand.
|
||||
const SCHEDULED_TASK_UNIT_BUDGET = 200;
|
||||
|
||||
// Cap on the per-tick list of configs blocked by an active run. Blocked
|
||||
// configs leave no durable trace on their row, so the summary names them.
|
||||
const ALREADY_RUNNING_IDS_CAP = 20;
|
||||
|
||||
// Cron body for the `scheduled` Worker handler: start a rank-check run for every
|
||||
// config that's due. Wrapped in `withPgClient` at the entrypoint (server.ts).
|
||||
export async function runScheduledRankChecks(env: Env) {
|
||||
const nowIso = new Date().toISOString();
|
||||
const dueConfigs =
|
||||
await RankTrackingRepository.getDueConfigsWithOrganization(nowIso);
|
||||
|
||||
const isHosted = await isHostedServerAuthMode();
|
||||
const keywordCounts = await RankTrackingRepository.getKeywordCountsForConfigs(
|
||||
dueConfigs.map((config) => config.id),
|
||||
);
|
||||
|
||||
// Function-local so it lives exactly one tick: at module scope this would be
|
||||
// cross-invocation global state in Workers, and a rejection would be cached
|
||||
// forever. Within a tick, a rejection staying memoized is intentional — one
|
||||
// Autumn call per org, and that org's configs simply stay due.
|
||||
const paidPlanChecks = new Map<string, Promise<boolean>>();
|
||||
const checkPaidPlan = (organizationId: string) => {
|
||||
let check = paidPlanChecks.get(organizationId);
|
||||
if (!check) {
|
||||
check = customerHasPaidPlan(organizationId, { retryDenied: true });
|
||||
paidPlanChecks.set(organizationId, check);
|
||||
}
|
||||
return check;
|
||||
};
|
||||
|
||||
let unitsStarted = 0;
|
||||
let started = 0;
|
||||
let stoppedByBudget = false;
|
||||
let skippedFree = 0;
|
||||
let skippedNoKeywords = 0;
|
||||
let concurrentChangeSkips = 0;
|
||||
let alreadyRunning = 0;
|
||||
const alreadyRunningConfigIds: string[] = [];
|
||||
let planCheckErrors = 0;
|
||||
let workflowStartErrors = 0;
|
||||
let configErrors = 0;
|
||||
|
||||
for (const config of dueConfigs) {
|
||||
// Per-config containment: one bad row (e.g. malformed next_check_at, which
|
||||
// sorts first and would head every scan) or transient DB error must not
|
||||
// starve the rest of the tick or suppress the summary log.
|
||||
try {
|
||||
// Skip configs whose org doesn't have a paid plan
|
||||
if (isHosted && !(await customerHasPaidPlan(config.organizationId))) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Skip configs with no keywords before advancing the schedule
|
||||
const kwCount = await RankTrackingRepository.getKeywordCountForConfig(
|
||||
config.id,
|
||||
);
|
||||
if (kwCount === 0) {
|
||||
console.log(
|
||||
`[cron] Skipping config ${config.id} (${config.domain}) — no keywords`,
|
||||
);
|
||||
// Still advance schedule so this config doesn't stay due forever
|
||||
const skipInterval = isScheduledRankTrackingInterval(
|
||||
config.scheduleInterval,
|
||||
)
|
||||
? config.scheduleInterval
|
||||
: null;
|
||||
if (skipInterval) {
|
||||
await RankTrackingRepository.updateConfig(
|
||||
config.id,
|
||||
config.projectId,
|
||||
{
|
||||
nextCheckAt: computeNextCheckAt(skipInterval, config.nextCheckAt),
|
||||
},
|
||||
);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Advance nextCheckAt immediately to prevent retry storms if the run fails
|
||||
const interval = isScheduledRankTrackingInterval(config.scheduleInterval)
|
||||
? config.scheduleInterval
|
||||
: null;
|
||||
if (interval) {
|
||||
await RankTrackingRepository.updateConfig(config.id, config.projectId, {
|
||||
nextCheckAt: computeNextCheckAt(interval, config.nextCheckAt),
|
||||
});
|
||||
// Unreachable: the due query excludes manual configs and NULL next check
|
||||
// times. Narrow rather than assert so a query change can't produce a run
|
||||
// with no schedule anchor.
|
||||
if (!interval || !config.nextCheckAt) continue;
|
||||
|
||||
const kwCount = keywordCounts.get(config.id) ?? 0;
|
||||
const taskUnits = kwCount * devicesCount(config.devices);
|
||||
// Projected stop: admit only what fits the budget. The first start of a
|
||||
// tick is exempt so an oversized config can never starve, and zero-unit
|
||||
// rows (no keywords) always advance.
|
||||
if (
|
||||
started > 0 &&
|
||||
unitsStarted + taskUnits > SCHEDULED_TASK_UNIT_BUDGET
|
||||
) {
|
||||
stoppedByBudget = true;
|
||||
break;
|
||||
}
|
||||
|
||||
const result = await beginRankCheckRun({
|
||||
workflow: env.RANK_CHECK_WORKFLOW,
|
||||
config,
|
||||
projectId: config.projectId,
|
||||
billingCustomer: {
|
||||
userId: "system",
|
||||
userEmail: "system@openseo.so",
|
||||
organizationId: config.organizationId,
|
||||
projectId: config.projectId,
|
||||
},
|
||||
keywordsTotal: kwCount,
|
||||
trigger: "scheduled",
|
||||
workflowStartErrorMessage: "Failed to start scheduled workflow",
|
||||
});
|
||||
const observedNextCheckAt = config.nextCheckAt;
|
||||
const nextCheckAt = computeNextCheckAt(interval, observedNextCheckAt);
|
||||
|
||||
if (!result.ok) {
|
||||
console.log(
|
||||
`[cron] Skipping config ${config.id} (${config.domain}) — run already active`,
|
||||
if (kwCount === 0) {
|
||||
const claimed = await RankTrackingRepository.claimDueConfig({
|
||||
configId: config.id,
|
||||
projectId: config.projectId,
|
||||
observedNextCheckAt,
|
||||
nextCheckAt,
|
||||
lastSkipReason: "no_keywords",
|
||||
});
|
||||
if (claimed) skippedNoKeywords++;
|
||||
else concurrentChangeSkips++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Self-hosted deployments treat every config as paid and make no Autumn
|
||||
// calls at all.
|
||||
let hasPaidPlan = true;
|
||||
if (isHosted) {
|
||||
try {
|
||||
hasPaidPlan = await checkPaidPlan(config.organizationId);
|
||||
} catch (err) {
|
||||
// Never write nextCheckAt on an error: it is the schedule anchor, so
|
||||
// an error write would permanently shift this config's slot and
|
||||
// herd-sync configs after an outage. Leaving the row due is the retry.
|
||||
console.error(
|
||||
`[cron] Plan check failed for config ${config.id} (${config.domain}):`,
|
||||
err,
|
||||
);
|
||||
planCheckErrors++;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if (!hasPaidPlan) {
|
||||
const claimed = await RankTrackingRepository.claimDueConfig({
|
||||
configId: config.id,
|
||||
projectId: config.projectId,
|
||||
observedNextCheckAt,
|
||||
nextCheckAt,
|
||||
lastSkipReason: "plan_required",
|
||||
});
|
||||
if (claimed) skippedFree++;
|
||||
else concurrentChangeSkips++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Claim the slot before starting. Clearing lastSkipReason here is what
|
||||
// lets an upgraded org drop the "plan_required" badge — the workflow only
|
||||
// writes null on a fully successful run.
|
||||
const claimed = await RankTrackingRepository.claimDueConfig({
|
||||
configId: config.id,
|
||||
projectId: config.projectId,
|
||||
observedNextCheckAt,
|
||||
nextCheckAt,
|
||||
lastSkipReason: null,
|
||||
});
|
||||
if (!claimed) {
|
||||
concurrentChangeSkips++;
|
||||
continue;
|
||||
}
|
||||
|
||||
let result;
|
||||
try {
|
||||
result = await beginRankCheckRun({
|
||||
workflow: env.RANK_CHECK_WORKFLOW,
|
||||
config,
|
||||
projectId: config.projectId,
|
||||
billingCustomer: {
|
||||
userId: "system",
|
||||
userEmail: "system@openseo.so",
|
||||
organizationId: config.organizationId,
|
||||
projectId: config.projectId,
|
||||
},
|
||||
keywordsTotal: kwCount,
|
||||
trigger: "scheduled",
|
||||
workflowStartErrorMessage: "Failed to start scheduled workflow",
|
||||
});
|
||||
} catch (err) {
|
||||
// Leave the schedule advanced: a systemic Workflows outage must not
|
||||
// make hundreds of configs due again on the next tick.
|
||||
workflowStartErrors++;
|
||||
console.error(
|
||||
`[cron] Failed to start scheduled rank check for config ${config.id} (${config.domain}):`,
|
||||
err,
|
||||
);
|
||||
} else {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (result.ok) {
|
||||
unitsStarted += taskUnits;
|
||||
started++;
|
||||
continue;
|
||||
}
|
||||
|
||||
alreadyRunning++;
|
||||
if (alreadyRunningConfigIds.length < ALREADY_RUNNING_IDS_CAP) {
|
||||
alreadyRunningConfigIds.push(config.id);
|
||||
}
|
||||
// Nothing was started, so give the slot back and retry next tick once the
|
||||
// blocking run clears. A manual edit landing in between wins the CAS.
|
||||
const restored = await RankTrackingRepository.claimDueConfig({
|
||||
configId: config.id,
|
||||
projectId: config.projectId,
|
||||
observedNextCheckAt: nextCheckAt,
|
||||
nextCheckAt: observedNextCheckAt,
|
||||
});
|
||||
if (!restored) {
|
||||
console.log(
|
||||
`[cron] Started scheduled rank check ${result.runId} for config ${config.id} (${config.domain})`,
|
||||
`[cron] Could not restore schedule for config ${config.id} (${config.domain}) — changed concurrently`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
configErrors++;
|
||||
console.error(
|
||||
`[cron] Error processing config ${config.id} (${config.domain}):`,
|
||||
err,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Oldest by the due query's next_check_at ASC ordering.
|
||||
const oldestDue = dueConfigs[0]?.nextCheckAt;
|
||||
// Object argument (not an interpolated string) so Workers Logs indexes the
|
||||
// fields. Error level when anything failed, so ticks that need attention
|
||||
// surface in error-filtered views.
|
||||
const logSummary =
|
||||
planCheckErrors + workflowStartErrors + configErrors > 0
|
||||
? console.error
|
||||
: console.log;
|
||||
logSummary({
|
||||
event: "rank_tracking_scheduler_summary",
|
||||
candidates: dueConfigs.length,
|
||||
started,
|
||||
unitsStarted,
|
||||
budget: SCHEDULED_TASK_UNIT_BUDGET,
|
||||
stoppedByBudget,
|
||||
skippedFree,
|
||||
skippedNoKeywords,
|
||||
concurrentChangeSkips,
|
||||
alreadyRunning,
|
||||
alreadyRunningConfigIds,
|
||||
planCheckErrors,
|
||||
workflowStartErrors,
|
||||
configErrors,
|
||||
oldestDueAgeMs: oldestDue
|
||||
? Date.now() - new Date(oldestDue).getTime()
|
||||
: null,
|
||||
});
|
||||
}
|
||||
|
||||
@ -91,4 +91,23 @@ describe("rank tracking schedules", () => {
|
||||
"2026-03-31T05:30:00.000Z",
|
||||
);
|
||||
});
|
||||
|
||||
it("preserves the time-of-day anchor for heavily overdue daily schedules", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-03-10T12:00:00.000Z"));
|
||||
|
||||
expect(computeNextCheckAt("daily", "2026-01-31T05:30:00.000Z")).toBe(
|
||||
"2026-03-11T05:30:00.000Z",
|
||||
);
|
||||
});
|
||||
|
||||
it("preserves the weekday and time anchor for heavily overdue weekly schedules", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-03-10T12:00:00.000Z"));
|
||||
|
||||
// 2026-01-31 is a Saturday; every advance lands on a Saturday.
|
||||
expect(computeNextCheckAt("weekly", "2026-01-31T05:30:00.000Z")).toBe(
|
||||
"2026-03-14T05:30:00.000Z",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@ -111,6 +111,13 @@ type ScheduledRankTrackingInterval = Exclude<
|
||||
"manual"
|
||||
>;
|
||||
|
||||
// Values written to rank_tracking_configs.last_skip_reason (free-form text in
|
||||
// the schema; this union keeps writers and UI comparisons in sync).
|
||||
export type RankTrackingSkipReason =
|
||||
| "plan_required"
|
||||
| "no_keywords"
|
||||
| "insufficient_credits";
|
||||
|
||||
export function estimateScheduledRankCheckCredits(
|
||||
keywordCount: number,
|
||||
devices: RankTrackingConfig["devices"],
|
||||
|
||||
@ -72,7 +72,7 @@
|
||||
},
|
||||
],
|
||||
"triggers": {
|
||||
"crons": ["*/15 * * * *"],
|
||||
"crons": ["*/5 * * * *"],
|
||||
},
|
||||
// This config serves local dev and Docker self-host only. All Cloudflare
|
||||
// deployments — previews, prod, self-host — go through Alchemy
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user