1/**
2 * Runs recurring cron jobs through cluster sharding.
3 *
4 * This module turns a `Cron.Cron` schedule into a `Layer` that coordinates one
5 * recurring job across a cluster. It registers a singleton for the initial
6 * scheduling step and a persisted entity message for each run. This is useful
7 * for distributed maintenance work where the job should be owned by the cluster
8 * rather than by every runner independently.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Cron from "../Cron.ts"
14import * as DateTime from "../DateTime.ts"
15import * as Duration from "../Duration.ts"
16import * as Effect from "../Effect.ts"
17import * as Exit from "../Exit.ts"
18import * as Layer from "../Layer.ts"
19import * as Option from "../Option.ts"
20import * as PrimaryKey from "../PrimaryKey.ts"
21import * as Rpc from "../rpc/Rpc.ts"
22import * as Schedule from "../Schedule.ts"
23import * as Schema from "../Schema.ts"
24import type { Scope } from "../Scope.ts"
25import * as ClusterSchema from "./ClusterSchema.ts"
26import { Persisted, Uninterruptible } from "./ClusterSchema.ts"
27import * as DeliverAt from "./DeliverAt.ts"
28import * as Entity from "./Entity.ts"
29import type { Sharding } from "./Sharding.ts"
30import * as Singleton from "./Singleton.ts"
31
32/**
33 * Creates a layer that runs a cron job through the cluster sharding system.
34 *
35 * **Details**
36 *
37 * The job is scheduled as persisted entity messages, with an initial singleton
38 * scheduling step and optional controls for shard group, next-run calculation,
39 * and skipping stale scheduled runs.
40 *
41 * @stability unstable
42 * @category constructors
43 * @since 4.0.0
44 */
45export const make = <E, R>(options: {
46 readonly name: string
47 readonly cron: Cron.Cron
48 readonly execute: Effect.Effect<void, E, R>
49
50 /**
51 * Choose a shard group to run this cron job on.
52 */
53 readonly shardGroup?: string | undefined
54
55 /**
56 * Controls whether the next cron job is based on the time of the previous
57 * run.
58 *
59 * **Details**
60 *
61 * Defaults to `false`, meaning the next run will be calculated from the
62 * current time.
63 */
64 readonly calculateNextRunFromPrevious?: boolean | undefined
65
66 /**
67 * If set, the cron job will skip execution if the scheduled time is older
68 * than this duration.
69 *
70 * **When to use**
71 *
72 * Use to prevent running jobs that were scheduled too far in the past.
73 *
74 * **Details**
75 *
76 * Defaults to "1 day".
77 */
78 readonly skipIfOlderThan?: Duration.Input | undefined
79}): Layer.Layer<never, never, Sharding | Exclude<R, Scope>> => {
80 const CronEntity = Entity.make(`ClusterCron/${options.name}`, [
81 Rpc.make("run", {
82 payload: CronPayload
83 })
84 .annotate(Persisted, true)
85 .annotate(Uninterruptible, true)
86 ])
87 .annotate(ClusterSchema.ShardGroup, () => options.shardGroup ?? "default")
88 .annotate(ClusterSchema.ClientTracingEnabled, false)
89
90 const InitialRun = Singleton.make(
91 `ClusterCron/${options.name}`,
92 Effect.gen(function*() {
93 const now = yield* DateTime.now
94 const next = DateTime.fromDateUnsafe(Cron.next(options.cron, now))
95 const entityId = options.calculateNextRunFromPrevious ? "initial" : DateTime.formatIso(next)
96 const client = (yield* CronEntity.client)(entityId)
97 yield* client.run({ dateTime: next }, { discard: true })
98 }),
99 { shardGroup: options.shardGroup }
100 )
101
102 const skipIfOlderThan = Option.fromUndefinedOr(options.skipIfOlderThan).pipe(
103 Option.map(Duration.fromInputUnsafe),
104 Option.getOrElse(() => Duration.days(1))
105 )
106
107 const effect = Effect.fnUntraced(function*(dateTime: DateTime.Utc) {
108 const now = yield* DateTime.now
109 if (DateTime.isLessThan(dateTime, DateTime.subtractDuration(now, skipIfOlderThan))) {
110 return
111 }
112 return yield* options.execute
113 }, Effect.orDie)
114
115 const EntityLayer = CronEntity.toLayer(Effect.gen(function*() {
116 const makeClient = yield* CronEntity.client
117 return {
118 run: (request) =>
119 Effect.onExitPrimitive(
120 effect(request.payload.dateTime),
121 Effect.fnUntraced(function*(exit) {
122 if (Exit.isFailure(exit)) {
123 yield* Effect.logWarning(exit.cause)
124 }
125 const now = yield* DateTime.now
126 const next = DateTime.fromDateUnsafe(Cron.next(
127 options.cron,
128 options.calculateNextRunFromPrevious ? request.payload.dateTime : now
129 ))
130 const client = makeClient(DateTime.formatIso(next))
131 return yield* client.run({ dateTime: next }, { discard: true }).pipe(
132 Effect.tapCause((cause) => Effect.logWarning("Failed to schedule next run, retrying", cause)),
133 Effect.sandbox,
134 Effect.retry(retryPolicy),
135 Effect.orDie
136 )
137 }),
138 true
139 ).pipe(
140 Effect.annotateLogs({
141 module: "effect/cluster/ClusterCron",
142 name: options.name,
143 dateTime: request.payload.dateTime
144 })
145 )
146 }
147 }))
148
149 return Layer.merge(InitialRun, EntityLayer)
150}
151
152const retryPolicy = Schedule.min([
153 Schedule.exponential(200, 1.5),
154 Schedule.spaced("1 minute")
155])
156
157class CronPayload extends Schema.Class<CronPayload>("effect/cluster/ClusterCron/CronPayload")({
158 dateTime: Schema.DateTimeUtc
159}) {
160 [PrimaryKey.symbol]() {
161 return ""
162 }
163 [DeliverAt.symbol]() {
164 return this.dateTime
165 }
166}