Inngest.ts

import * as Http from "@effect/platform/HttpClient"
import { Schema } from "@effect/schema"
import { Array, Config, Context, Data, Effect, Layer, flow } from "effect"
import { Inngest as Inngest_ } from "inngest"

class InngestRun extends Schema.Class<InngestRun>("InngestRun")({
  run_id: Schema.String,
  status: Schema.Literal("Running", "Completed", "Failed", "Cancelled"),
}) {
  static decodeArray = Http.response.schemaBodyJsonScoped(
    Schema.Array(InngestRun),
  )
}

class InngestError extends Data.TaggedError("InngestError")<{
  readonly error: unknown
}> {}

const make = Effect.gen(function* () {
  const config = yield* Config.all({
    id: Config.string("INNGEST_APP_ID"),
    signingKey: Config.string("INNGEST_SIGNING_KEY"),
  })

  const inngest = new Inngest_({ id: config.id })

  const client = (yield* Http.client.Client).pipe(
    Http.client.mapRequest(
      flow(
        Http.request.bearerToken(config.signingKey),
        Http.request.prependUrl("https://api.inngest.com/v1"),
      ),
    ),
  )

  const use = <A>(
    f: (client: Inngest_) => Promise<A>,
  ): Effect.Effect<A, InngestError> =>
    Effect.tryPromise({
      try: () => f(inngest),
      catch: error => new InngestError({ error }),
    })

  const runs = (eventId: string) =>
    client(Http.request.get(`/events/${eventId}/runs`)).pipe(
      InngestRun.decodeArray,
    )

  const firstRun = (eventId: string) =>
    runs(eventId).pipe(Effect.flatMap(Array.head))

  const waitUntilFinished = (eventId: string) =>
    firstRun(eventId).pipe(
      Effect.delay(1000),
      Effect.repeat({ until: run => run.status !== "Running" }),
    )

  return { runs, use, waitUntilFinished } as const
})

export class Inngest extends Context.Tag("app/Inngest")<
  Inngest,
  Effect.Effect.Success<typeof make>
>() {
  static Live = Layer.effect(Inngest, make).pipe(
    Layer.provide(Http.client.layer),
  )
}