import { withTimeLog } from "https://esm.town/v/dvdsgl/effect"; import { Chunk, Context, Data, Effect, Layer, Logger, LogLevel, Option, Stream } from "npm:effect"; import { UnknownException } from "npm:effect/Cause"; import { v4 as uuidv4 } from "npm:uuid"; const baseUrl = `https://api.glideapps.com`; type Stash = { $stashID: string }; type PutResult = { data: { rowIDs: string[] } }; export type GlideInfo = { token: string }; export class Glide extends Context.Tag("Glide")() {} export class GlideError extends Data.TaggedError("GlideError")<{ type?: string; message: string; }> {} export const layer = (info: GlideInfo) => Layer.succeed(Glide, Glide.of(info)); const getGlide: Effect.Effect = Effect.gen(function*() { const provided = yield* Effect.serviceOption(Glide); if (Option.isSome(provided)) return provided.value; const env = { token: Deno.env.get("glide"), }; if (typeof env.token === "string") return env; yield* new GlideError({ message: "No Glide layer provided, and 'glide' secret not found in environment", }); }); function removeUnsupportedColumnTypes(row: any): any { const pr = { ...row }; for (const [k, v] of Object.entries(pr)) { // Glide does not accept arrays or objects yet if (Array.isArray(v) || typeof v === "object") { delete pr[k]; } } return pr; } // Handles calling fetch, checking for API errors function fetchGlide( method: "GET" | "POST" | "PUT", url: string, body: unknown, ): Effect.Effect { return Effect.gen(function*() { const { token } = yield* getGlide; const result = yield* Effect.tryPromise(() => fetch(url, { method, headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json", }, body: JSON.stringify(body), }) ).pipe( Effect.tap(Effect.annotateCurrentSpan({ method, url })), Effect.withLogSpan("fetchGlide"), ); if (!result.ok) { yield* new GlideError({ message: result.statusText }); } const json = yield* Effect.tryPromise(() => result.json()); if ("error" in json) { yield* new GlideError(json.error); } return json; }).pipe( Effect.catchTag( "UnknownException", () => new GlideError({ message: "Could not fetch" }), ), ); } export const putTable = ( id: string, rows: Stash | any[], ): Effect.Effect => Effect.gen(function*() { // Some tables have a prefix, e.g. "native-table-" but this API // does not accept it. id = id.replace("native-table-", ""); const url = `${baseUrl}/tables/${id}?onSchemaError=updateSchema`; yield* Effect.logDebug(`putTable ${url}`); if (Array.isArray(rows)) { rows = rows.map(removeUnsupportedColumnTypes); } const results = yield* fetchGlide("PUT", url, { rows, schema: { columns: [] }, }); yield* Effect.logDebug("putTable succeeded"); return results; }).pipe(withTimeLog(`putTable`)); export const makeStash = (): Stash => ({ $stashID: uuidv4() }); export const postStash = ( stash: string | Stash, rows: any[], serial: string | number, ): Effect.Effect => Effect.gen(function*() { const id = typeof stash === "string" ? stash : stash.$stashID; rows = rows.map(removeUnsupportedColumnTypes); const url = `${baseUrl}/stashes/${id}/${serial}`; yield* Effect.logDebug(`postStash ${serial}`); yield* fetchGlide("POST", url, rows); return { $stashID: id }; }).pipe( // Sometimes the API just fails Effect.retry({ times: 3 }), ); type Import = { table: string; getRows: () => AsyncGenerator; }; async function* bufferAsyncIterable( asyncIterable: AsyncIterable, max = 1_000, ): AsyncGenerator { let buffer: T[] = []; for await (let chunk of asyncIterable) { if (!Array.isArray(chunk)) chunk = [chunk]; buffer.push(...chunk); while (buffer.length >= max) yield buffer.splice(0, max); } // Yield any remaining elements in the buffer if (buffer.length > 0) yield buffer; } export function importTableEffect(props: Import) { const { table, getRows } = props; return Effect.gen(function*() { yield* Effect.logDebug(`ingest start`); const stash = makeStash(); const stream = Stream.fromAsyncIterable( // We buffer the rows into larger chunks // because many APIs only fetch small chunks (e.g. 100) // records at a time. bufferAsyncIterable(getRows()), (e) => new Error(String(e)), ); let totalRows = 0; let serial = 0; yield* Stream.runForEach(stream, (rows) => Effect.gen(function*() { totalRows += rows.length; yield* Effect.logDebug(`ingest ${totalRows} total rows`); yield* postStash(stash, rows, serial++); })); return yield* putTable(table, stash); }).pipe(withTimeLog(`ingest`)); } // Runs an injestion job. export async function importTable( props: Import & { debug?: boolean }, ) { const program = importTableEffect(props).pipe( Logger.withMinimumLogLevel(props.debug ? LogLevel.Debug : LogLevel.Info), ); return await Effect.runPromise(program); }