feat(sidetrack): allow initializing sidetrack on layer creation

Author: aniravi24

.changeset/fresh-squids-push.md

@@ -0,0 +1,5 @@
+---
+"sidetrack": patch
+---
+
+Support initializing sidetrack on layer creation

packages/sidetrack/src/effect.ts

@@ -504,7 +504,7 @@ export function layer<Queues extends SidetrackQueuesGenericType>(
                 `INSERT INTO sidetrack_cron_jobs (queue, cron_expression, payload, timezone)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (queue, cron_expression) DO UPDATE
-           SET payload = $3 RETURNING *`,
+           SET payload = $3, timezone = $4 RETURNING *`,
                 [
                   queueName,
                   cronExpression,
@@ -770,6 +770,10 @@ export function layer<Queues extends SidetrackQueuesGenericType>(
           }),
         );
 
+      if (layerOptions.startOnInitialization) {
+        yield* start();
+      }
+
       return {
         cancelJob,
         deactivateCronSchedule,

packages/sidetrack/src/types.ts

@@ -83,6 +83,12 @@ export interface SidetrackOptions<Queues extends SidetrackQueuesGenericType> {
    */
   pollingInterval?: PollingInterval;
   queues: SidetrackQueues<Queues>;
+  /**
+   * This is useful if you want it to automatically start polling when you initialize sidetrack instead of waiting to call start.
+   * Particularly useful if you are running sidetrack in a long running process on its own.
+   * e.g. if you are using Layer.launch with Effect (@link https://effect.website/docs/requirements-management/layers/#converting-a-layer-to-an-effect)
+   */
+  startOnInitialization?: boolean;
 }
 
 export class SidetrackJobRunError extends Data.TaggedError(

packages/sidetrack/test/effect.test.ts

@@ -454,4 +454,51 @@ describe("Effect API", () => {
       await Effect.runPromise(program.pipe(Effect.provide(sidetrackLayer)));
     });
   });
+
+  it("automatically starts job processing when startOnInitialization is true", async () => {
+    await runInTransaction(pool, async (client) => {
+      interface Queues {
+        test: { id: string };
+      }
+
+      const SidetrackService = SidetrackEffect.getSidetrackService<Queues>();
+      let jobProcessed = false;
+
+      const sidetrackLayer = SidetrackEffect.layer<Queues>({
+        dbClient: usePg(client),
+        pollingInterval: 100,
+        queues: {
+          test: {
+            run: (payload) => {
+              jobProcessed = true;
+              expect(payload.id).toBe("auto-start");
+              return Promise.resolve(payload);
+            },
+          },
+        },
+        startOnInitialization: true,
+      });
+
+      const program = Effect.gen(function* () {
+        const sidetrack = yield* SidetrackService;
+
+        // Insert a job that should be automatically processed
+        const job = yield* sidetrack.insertJob("test", { id: "auto-start" });
+
+        // Allow some time for the job to be processed by the automatic polling
+        yield* Effect.sleep("300 millis");
+
+        // We shouldn't need to call start() or runJob() explicitly
+        const completedJob = yield* sidetrack.getJob(job.id);
+
+        expect(jobProcessed).toBe(true);
+        expect(completedJob.status).toBe("completed");
+
+        // Clean up by stopping the service
+        sidetrack.stop();
+      });
+
+      await Effect.runPromise(program.pipe(Effect.provide(sidetrackLayer)));
+    });
+  });
 });