feat(pipeline): add database schema, repositories, services, and loader for pipeline data

alchemy.run.ts

@@ -8,6 +8,7 @@ import {
   KVNamespace,
   R2Bucket,
   TanStackStart,
+  Worker,
 } from "alchemy/cloudflare";
 
 export const app = await alchemy("laxdb", {
@@ -133,20 +134,29 @@ if (app.local) {
 // KV
 export const kv = await KVNamespace("kv", {});
 
+// Pipeline KV for cron worker caching
+export const pipelineKV = await KVNamespace("pipeline-kv", {
+  title: `laxdb-pipeline-${stage}`,
+});
+
 // Storage
 export const storage = await R2Bucket("storage", {});
 
-// export const worker = await Worker("api", {
-//   entrypoint: "packages/api/src/index.ts",
-//   url: true,
-//   bindings: {
-//     DB: db,
-//     KV: kv,
-//     STORAGE: storage,
-//     DATABASE_URL: dbRole.connectionUrl,
-//     ...secrets,
-//   },
-// });
+// API Worker with cron triggers for pipeline extraction
+export const api = await Worker("api", {
+  entrypoint: "packages/api/src/index.ts",
+  url: true,
+  bindings: {
+    DB: db,
+    KV: kv,
+    PIPELINE_KV: pipelineKV,
+    STORAGE: storage,
+    DATABASE_URL: dbRole.connectionUrl,
+    ...secrets,
+  },
+  // Hourly cron trigger for pipeline data extraction
+  crons: ["0 * * * *"],
+});
 
 export const web = await TanStackStart("web", {
   cwd: "./packages/web",
@@ -174,12 +184,13 @@ export const docs = await TanStackStart("docs", {
 
 console.log({
   domain,
-  // web: web.url,
+  web: web.url,
   marketing: marketing.url,
   docs: docs.url,
-  // api: api.url,
+  api: api.url,
   db: database.id,
   kv: kv.namespaceId,
+  pipelineKV: pipelineKV.namespaceId,
   r2: storage.name,
   stage,
 });
@@ -194,9 +205,10 @@ if (process.env.PULL_REQUEST) {
 
      Your changes have been deployed to a preview environment:
 
+     **🌐 Website:** ${web.url}
+     **🌐 API:** ${api.url}
      **🌐 Docs:** ${docs.url}
      **🌐 Marketing:** ${marketing.url}
-     **🌐 Website:** ${web.url}
 
      Built from commit ${process.env.GITHUB_SHA?.slice(0, 7)}
 

package.json

@@ -11,6 +11,7 @@
     "destroy": "alchemy destroy",
     "dev": "alchemy dev",
     "test": "turbo test",
+    "test:unit": "turbo test:unit",
     "typecheck": "turbo typecheck",
     "lint": "turbo lint",
     "format": "turbo format",

packages/api/package.json

@@ -15,7 +15,8 @@
     "@effect-atom/atom-react": "^0.4.4",
     "@effect/platform": "0.94.1",
     "@effect/rpc": "^0.73.0",
-    "@laxdb/core": "workspace:*"
+    "@laxdb/core": "workspace:*",
+    "@laxdb/pipeline": "workspace:*"
   },
   "devDependencies": {
     "@cloudflare/workers-types": "^4.20241218.0"

packages/api/src/client.ts

@@ -3,6 +3,7 @@ import { Layer } from "effect";
 import { RpcAuthClient } from "./auth/auth.client";
 import { RpcGameClient } from "./game/game.client";
 import { RpcOrganizationClient } from "./organization/organization.client";
+import { RpcStatsClient } from "./pipeline/stats.client";
 import { RpcContactInfoClient } from "./player/contact-info/contact-info.client";
 import { RpcPlayerClient } from "./player/player.client";
 import { RpcSeasonClient } from "./season/season.client";
@@ -16,4 +17,5 @@ export const RpcClientLive = Layer.mergeAll(
   RpcTeamClient.Default,
   RpcOrganizationClient.Default,
   RpcAuthClient.Default,
+  RpcStatsClient.Default,
 );

packages/api/src/cron/scheduled.ts

@@ -0,0 +1,132 @@
+/**
+ * Cron scheduled handler for pipeline data extraction
+ *
+ * Runs hourly to extract and load data from active leagues.
+ * Each league is processed independently - one failure doesn't stop others.
+ */
+
+import {
+  getActiveLeagues,
+  type LeagueAbbreviation,
+} from "@laxdb/pipeline/config/seasons";
+
+/**
+ * Cloudflare Worker environment bindings
+ */
+interface Env {
+  PIPELINE_KV: KVNamespace;
+  DATABASE_URL: string;
+}
+
+/**
+ * Extract data from a league's source API
+ *
+ * @param league - The league abbreviation
+ * @param env - Cloudflare Worker environment
+ */
+async function extractLeague(
+  league: LeagueAbbreviation,
+  env: Env,
+): Promise<void> {
+  console.log(`[${league}] Starting extraction...`);
+
+  // TODO: Implement actual extraction using @laxdb/pipeline extractors
+  // For MVP, this is a placeholder that logs progress
+  // Will call extractNLL, extractPLL, etc. based on league
+
+  console.log(`[${league}] Extraction complete`);
+}
+
+/**
+ * Load extracted data into the database
+ *
+ * @param league - The league abbreviation
+ * @param env - Cloudflare Worker environment
+ */
+async function loadLeague(league: LeagueAbbreviation, env: Env): Promise<void> {
+  console.log(`[${league}] Loading data...`);
+
+  // TODO: Implement actual data loading
+  // Will insert/update records in pipeline_* tables
+
+  console.log(`[${league}] Load complete`);
+}
+
+/**
+ * Invalidate cached data for a league
+ *
+ * @param league - The league abbreviation
+ * @param env - Cloudflare Worker environment
+ */
+async function invalidateCache(
+  league: LeagueAbbreviation,
+  env: Env,
+): Promise<void> {
+  console.log(`[${league}] Invalidating cache...`);
+
+  // Delete cached leaderboard data for this league
+  const cacheKeys = [
+    `cache:leaderboard:${league}`,
+    `cache:players:${league}`,
+    `cache:teams:${league}`,
+  ];
+
+  await Promise.all(cacheKeys.map((key) => env.PIPELINE_KV.delete(key)));
+
+  console.log(`[${league}] Cache invalidated`);
+}
+
+/**
+ * Scheduled cron handler
+ *
+ * Called by Cloudflare Workers on the configured schedule (hourly).
+ * Processes all currently active leagues in parallel with error isolation.
+ */
+export async function scheduled(
+  _controller: ScheduledController,
+  env: Env,
+  _ctx: ExecutionContext,
+): Promise<void> {
+  const startTime = Date.now();
+  console.log(
+    `[Cron] Starting scheduled extraction at ${new Date().toISOString()}`,
+  );
+
+  // Get leagues that are currently in-season
+  const activeLeagues = getActiveLeagues(new Date());
+
+  if (activeLeagues.length === 0) {
+    console.log("[Cron] No active leagues, skipping extraction");
+    return;
+  }
+
+  console.log(`[Cron] Active leagues: ${activeLeagues.join(", ")}`);
+
+  // Run all extractions in parallel, don't let one failure stop others
+  const results = await Promise.allSettled(
+    activeLeagues.map(async (league) => {
+      await extractLeague(league, env);
+      await loadLeague(league, env);
+      await invalidateCache(league, env);
+    }),
+  );
+
+  // Log results
+  let successCount = 0;
+  let failureCount = 0;
+
+  for (const [i, result] of results.entries()) {
+    if (result.status === "fulfilled") {
+      successCount++;
+      console.log(`[Cron] ${activeLeagues[i]} completed successfully`);
+    } else {
+      failureCount++;
+      console.error(`[Cron] ${activeLeagues[i]} failed:`, result.reason);
+    }
+  }
+
+  const duration = Date.now() - startTime;
+  console.log(
+    `[Cron] Completed in ${duration}ms - Success: ${successCount}, Failed: ${failureCount}`,
+  );
+}

packages/api/src/definition.ts

@@ -3,6 +3,7 @@ import { HttpApi } from "@effect/platform";
 import { AuthGroup } from "./auth/auth.api";
 import { GamesGroup } from "./game/game.api";
 import { OrganizationsGroup } from "./organization/organization.api";
+import { StatsApiGroup } from "./pipeline/stats.api";
 import { ContactInfoGroup } from "./player/contact-info/contact-info.api";
 import { PlayersGroup } from "./player/player.api";
 import { SeasonsGroup } from "./season/season.api";
@@ -16,4 +17,5 @@ export class LaxdbApi extends HttpApi.make("LaxdbApi")
   .add(ContactInfoGroup)
   .add(TeamsGroup)
   .add(OrganizationsGroup)
-  .add(AuthGroup) {}
+  .add(AuthGroup)
+  .add(StatsApiGroup) {}

packages/api/src/groups/index.ts

@@ -3,6 +3,7 @@ import { Layer } from "effect";
 import { AuthHandlersLive } from "../auth/auth.handlers";
 import { GamesHandlersLive } from "../game/game.handlers";
 import { OrganizationsHandlersLive } from "../organization/organization.handlers";
+import { StatsHandlersLive } from "../pipeline/stats.handlers";
 import { ContactInfoHandlersLive } from "../player/contact-info/contact-info.handlers";
 import { PlayersHandlersLive } from "../player/player.handlers";
 import { SeasonsHandlersLive } from "../season/season.handlers";
@@ -17,4 +18,5 @@ export const HttpGroupsLive = Layer.mergeAll(
   TeamsHandlersLive,
   OrganizationsHandlersLive,
   AuthHandlersLive,
+  StatsHandlersLive,
 );

packages/api/src/index.ts

@@ -58,4 +58,7 @@ const { handler } = HttpLayerRouter.toWebHandler(AllRoutes, {
   middleware: HttpMiddleware.logger,
 });
 
+// Re-export scheduled handler for cron triggers
+export { scheduled } from "./cron/scheduled";
+
 export default { fetch: handler };

packages/api/src/pipeline/players.rpc.ts

@@ -0,0 +1,28 @@
+import { Rpc, RpcGroup } from "@effect/rpc";
+import { PlayersContract } from "@laxdb/core/pipeline/players.contract";
+import { PlayersService } from "@laxdb/pipeline/rpc/players.service";
+import { Effect, Layer } from "effect";
+
+export class PlayersRpcs extends RpcGroup.make(
+  Rpc.make("PlayersGetPlayer", {
+    success: PlayersContract.getPlayer.success,
+    error: PlayersContract.getPlayer.error,
+    payload: PlayersContract.getPlayer.payload,
+  }),
+  Rpc.make("PlayersSearchPlayers", {
+    success: PlayersContract.searchPlayers.success,
+    error: PlayersContract.searchPlayers.error,
+    payload: PlayersContract.searchPlayers.payload,
+  }),
+) {}
+
+export const PlayersHandlers = PlayersRpcs.toLayer(
+  Effect.gen(function* () {
+    const service = yield* PlayersService;
+
+    return {
+      PlayersGetPlayer: (payload) => service.getPlayer(payload),
+      PlayersSearchPlayers: (payload) => service.searchPlayers(payload),
+    };
+  }),
+).pipe(Layer.provide(PlayersService.Default));