diff --git a/rivetkit-typescript/packages/effect/src/internal/ActionDispatcher.ts b/rivetkit-typescript/packages/effect/src/internal/ActionDispatcher.ts index eae860841c..9c000eb8cf 100644 --- a/rivetkit-typescript/packages/effect/src/internal/ActionDispatcher.ts +++ b/rivetkit-typescript/packages/effect/src/internal/ActionDispatcher.ts @@ -82,13 +82,19 @@ export const make = < ] as ( envelope: ActionRequest, ) => Action.ResultFrom; - // Raw RivetKit clients call no-argument actions with an - // absent first argument. The Effect JSON Void codec expects - // null, so adapt only actions that declared no payload. - const payloadForDecode = - !action.hasPayload && payload === undefined - ? null - : payload; + // RivetKit appends schedule metadata after positional arguments. + // For actions without payloads, this metadata occupies the payload + // slot. Effect's JSON Void codec expects that slot to contain null. + const actionExpectsVoid = !action.hasPayload; + const payloadWasOmitted = payload === undefined; + const scheduleMetadataIsPayload = + actionExpectsVoid && + meta === undefined && + isScheduledFireInfo(payload); + const shouldDecodeAsVoid = + (actionExpectsVoid && payloadWasOmitted) || + scheduleMetadataIsPayload; + const payloadForDecode = shouldDecodeAsVoid ? null : payload; const decodedPayload = yield* decodePayload( payloadForDecode, ).pipe( @@ -183,6 +189,17 @@ export const make = < ]; }); +const ScheduledFireInfoSchema: Schema.Schema = + Schema.Struct({ + kind: Schema.Literals(["at", "cron", "every"]), + id: Schema.String, + name: Schema.optionalKey(Schema.UndefinedOr(Schema.String)), + scheduledAt: Schema.Finite, + firedAt: Schema.Finite, + }); + +const isScheduledFireInfo = Schema.is(ScheduledFireInfoSchema); + const makeActorAbortedError = () => new Rivetkit.RivetError("actor", "aborted", "Actor aborted", { public: true, diff --git a/rivetkit-typescript/packages/effect/test/e2e.test.ts b/rivetkit-typescript/packages/effect/test/e2e.test.ts index 3bb7dced1a..30810e29a5 100644 --- a/rivetkit-typescript/packages/effect/test/e2e.test.ts +++ b/rivetkit-typescript/packages/effect/test/e2e.test.ts @@ -21,6 +21,8 @@ import { Pinger, PingerLive, ScaledOverflowError, + ScheduledNoPayload, + ScheduledNoPayloadLive, Strict, StrictLive, TransformedStateActor, @@ -81,6 +83,7 @@ const TestLayer = ReadyForEnvoy.pipe( WakeDecodeFailLive, BuildSetRejectedLive, TransformedStateActorLive, + ScheduledNoPayloadLive, ), ), Layer.provideMerge(Flags.layer), @@ -135,6 +138,49 @@ layer(TestLayer)("end-to-end", (it) => { }), ); + it.effect("runs scheduled actions without payloads", () => + Effect.gen(function* () { + const key = "t-scheduled-no-payload"; + const actor = (yield* ScheduledNoPayload.client).getOrCreate(key); + const flags = yield* Flags; + const firedFlag = `scheduled-no-payload:${key}`; + + yield* actor.ScheduleExpiration({ delay: 25 }); + + const fired = yield* Effect.sync(() => flags.get(firedFlag)).pipe( + Effect.repeat({ + until: (value) => value === true, + schedule: Schedule.spaced("100 millis"), + }), + Effect.timeout("5 seconds"), + TestClock.withLive, + ); + assert.strictEqual(fired, true); + assert.strictEqual(yield* actor.GetExpirationCount(), 1); + + const client = yield* Effect.acquireRelease( + Effect.sync(() => createClient({ endpoint, token, namespace })), + (client) => Effect.promise(() => client.dispose()), + ); + const rawActor = client.ScheduledNoPayload.getOrCreate( + "t-invalid-no-payload", + ); + const error = yield* Effect.promise(async () => { + try { + await Reflect.apply(rawActor.Expire, rawActor, [ + { unexpected: true }, + ]); + throw new Error("expected arbitrary payload to fail"); + } catch (error) { + return RawRivetErrors.toRivetError(error); + } + }); + + assert.strictEqual(error.group, "request"); + assert.strictEqual(error.code, "invalid"); + }), + ); + it.effect("rejects malformed raw action payloads as request.invalid", () => Effect.gen(function* () { const client = yield* Effect.acquireRelease( diff --git a/rivetkit-typescript/packages/effect/test/fixtures/actors.ts b/rivetkit-typescript/packages/effect/test/fixtures/actors.ts index 06da8cf362..9e2d1ae02c 100644 --- a/rivetkit-typescript/packages/effect/test/fixtures/actors.ts +++ b/rivetkit-typescript/packages/effect/test/fixtures/actors.ts @@ -526,6 +526,52 @@ export const CounterLive = Counter.toLayer( }), }, ); +// --- ScheduledNoPayload --- + +export const ScheduleExpiration = Action.make("ScheduleExpiration", { + payload: { delay: Schema.Number }, + success: Schema.String, +}); + +export const Expire = Action.make("Expire"); + +export const GetExpirationCount = Action.make("GetExpirationCount", { + success: Schema.Number, +}); + +export const ScheduledNoPayload = Actor.make("ScheduledNoPayload", { + actions: [ScheduleExpiration, Expire, GetExpirationCount], +}); + +export const ScheduledNoPayloadLive = ScheduledNoPayload.toLayer( + ({ rawRivetkitContext }) => + Effect.gen(function* () { + const expirationCount = yield* Ref.make(0); + const flags = yield* Flags; + const address = yield* Actor.CurrentAddress; + const firedFlag = `scheduled-no-payload:${address.key.join("/")}`; + + return ScheduledNoPayload.of({ + ScheduleExpiration: ({ payload }) => + Effect.promise(() => + rawRivetkitContext.schedule.after( + payload.delay, + "Expire", + ), + ), + Expire: () => + Ref.update(expirationCount, (count) => count + 1).pipe( + Effect.tap(() => + Effect.sync(() => { + flags.set(firedFlag, true); + }), + ), + Effect.asVoid, + ), + GetExpirationCount: () => Ref.get(expirationCount), + }); + }), +); // --- Strict ---