add worker create event and new query methods (#3524)

src/backend/clients/event/types.ts

@@ -332,6 +332,12 @@ export type EventMap = {
         charges: UsageInput[];
     };
 
+    // ---- Workers ----
+    // Only a genuinely new worker. Redeploying an existing name updates its
+    // row instead, and never reaches here — so a listener that prices this
+    // is pricing workers that came into existence, not deploys.
+    'worker.create': { actor: Actor; workerName: string };
+
     // ---- Outer / GUI broadcast ----
     'outer.cacheUpdate': {
         cacheKey: string[];

src/backend/drivers/workers/WorkerDriver.cloudflare.test.ts

@@ -47,7 +47,10 @@ import { PuterServer } from '../../server.js';
 import type { FSEntry } from '../../stores/fs/FSEntry.js';
 import { setupTestServer } from '../../testUtil.js';
 import { generateDefaultFsentries } from '../../util/userProvisioning.js';
-import type { WorkerDriver } from './WorkerDriver.js';
+import {
+    INTERNAL_ADMISSION_BYPASS,
+    type WorkerDriver,
+} from './WorkerDriver.js';
 
 const ACCOUNT_ID = 'cf-account';
 const AUTH_KEY = 'cf-auth-key';
@@ -282,6 +285,50 @@ describe('WorkerDriver.create with a configured deploy backend', () => {
         expect(Number(row!.root_dir_id)).toBe(secondEntry.id);
     });
 
+    // Anything pricing workers keys off this event, so a redeploy leaking one
+    // through would bill the user again for a worker they already own.
+    it('announces `worker.create` for a new worker but not for a redeploy', async () => {
+        const { user, actor } = await makeUser();
+        const path = `/${user.username}/worker.js`;
+        await writeSource(actor, user.id, path, 'v1');
+        const name = `announce-${user.username}`;
+
+        const announced: string[] = [];
+        const listener = (_key: unknown, data: { workerName: string }) => {
+            announced.push(data.workerName);
+        };
+        server.clients.event.on(
+            'worker.create',
+            listener as Parameters<typeof server.clients.event.on>[1],
+        );
+
+        try {
+            await inCtx(actor, () =>
+                target.create({ appId: '', workerName: name, filePath: path }),
+            );
+            await inCtx(actor, () =>
+                target.create({ appId: '', workerName: name, filePath: path }),
+            );
+            // The rehydrate shape: an existing worker redeployed past the
+            // admission gates because it is already ours.
+            await inCtx(actor, () =>
+                target.create({
+                    appId: '',
+                    workerName: name,
+                    filePath: path,
+                    [INTERNAL_ADMISSION_BYPASS]: true,
+                }),
+            );
+        } finally {
+            server.clients.event.off(
+                'worker.create',
+                listener as Parameters<typeof server.clients.event.off>[1],
+            );
+        }
+
+        expect(announced).toEqual([name]);
+    });
+
     it('rejects a name already taken by another user with 409', async () => {
         const owner = await makeUser();
         const stranger = await makeUser();

src/backend/drivers/workers/WorkerDriver.ts

@@ -292,6 +292,15 @@ export class WorkerDriver extends PuterDriver {
                 appOwner: appOwnerId,
                 preambleVersion,
             });
+            // Announced against the row rather than the deploy: the row is
+            // what makes the worker ours to keep, and it outlives a failed
+            // deploy. Awaited so a listener has settled before the caller is
+            // told the worker exists; one that throws is logged and ignored.
+            await this.clients.event.emitAndWait(
+                'worker.create',
+                { actor, workerName },
+                {},
+            );
         }
 
         // AppData is keyed by the app the worker authenticates as, so the

src/backend/stores/subdomain/SubdomainStore.test.ts

@@ -201,6 +201,34 @@ describe('SubdomainStore app_owner filtering', () => {
             await store.countByUserIdAndPrefix(userId, prefix, { appIds: [] }),
         ).toBe(0);
     });
+
+    it('counts only rows older than `createdBefore`', async () => {
+        const userId = await makeUser();
+        const prefix = `sds-aged-${Math.random().toString(36).slice(2, 6)}.`;
+        const old = await seed(userId, prefix, null);
+        await seed(userId, prefix, null);
+
+        // `ts` defaults to now for both, so one is backdated to put the two
+        // on opposite sides of the cutoff.
+        await server.clients.db.write(
+            'UPDATE `subdomains` SET `ts` = ? WHERE `subdomain` = ?',
+            ['2020-01-15 12:00:00', old],
+        );
+
+        expect(
+            await store.countByUserIdAndPrefix(userId, prefix, {
+                createdBefore: '2020-02-01 00:00:00',
+            }),
+        ).toBe(1);
+        // Older than everything, including the backdated row.
+        expect(
+            await store.countByUserIdAndPrefix(userId, prefix, {
+                createdBefore: '2019-01-01 00:00:00',
+            }),
+        ).toBe(0);
+        // Omitting it stays unfiltered.
+        expect(await store.countByUserIdAndPrefix(userId, prefix)).toBe(2);
+    });
 });
 
 // ── Listing, counting and scoping ───────────────────────────────────

src/backend/stores/subdomain/SubdomainStore.ts

@@ -393,7 +393,18 @@ export class SubdomainStore extends PuterStore {
     async countByUserIdAndPrefix(
         userId: number,
         prefix: string,
-        extra: { appId?: number; appIds?: number[] } = {},
+        extra: {
+            appId?: number;
+            appIds?: number[];
+            /**
+             * `YYYY-MM-DD HH:MM:SS`, matching rows strictly older than it.
+             * Compared in SQL because `ts` reaches JS as a string on one
+             * backend and a Date on another, and the string form parses as
+             * local time — a boundary computed here would drift by the server's
+             * offset.
+             */
+            createdBefore?: string;
+        } = {},
     ): Promise<number> {
         if (!userId || prefix == null) return 0;
 
@@ -404,6 +415,10 @@ export class SubdomainStore extends PuterStore {
             conditions.push(appOwnerFilter.condition);
             values.push(...appOwnerFilter.values);
         }
+        if (extra.createdBefore !== undefined) {
+            conditions.push('`ts` < ?');
+            values.push(extra.createdBefore);
+        }
         const rows = await this.clients.db.read(
             `SELECT COUNT(*) AS n FROM \`subdomains\` WHERE ${conditions.join(' AND ')}`,
             values,