Repository Pattern

The Repository Pattern provides an abstraction layer between the domain logic (services) and data persistence (database), located in worker/services/{domain}/repository.ts.

Purpose

Repositories handle:

Database queries and mutations
Data access logic
SQL query composition
Object-relational mapping
No business logic (that belongs in services)

Standard Repository Structure

export class FundRepository {
  constructor(private db: ProductionDBClient) {}

  // Create
  async create(data: CreateFundRequest): Promise<FundEntity> {
    const [fund] = await this.db
      .insert(fundsTable)
      .values({
        name: data.name,
        code: data.code,
        organizationId: data.organizationId,
        startDate: data.startDate,
        fees: data.fees ? JSON.stringify(data.fees) : null,
      })
      .returning();

    return this.mapToEntity(fund);
  }

  // Read
  async findById(id: number): Promise<FundEntity | null> {
    const [fund] = await this.db
      .select()
      .from(fundsTable)
      .where(eq(fundsTable.id, id));

    return fund ? this.mapToEntity(fund) : null;
  }

  // Update
  async update(id: number, data: UpdateFundRequest): Promise<FundEntity> {
    const [fund] = await this.db
      .update(fundsTable)
      .set({
        ...(data.name && { name: data.name }),
        ...(data.fees && { fees: JSON.stringify(data.fees) }),
        updatedAt: new Date(),
      })
      .where(eq(fundsTable.id, id))
      .returning();

    return this.mapToEntity(fund);
  }

  // Delete
  async delete(id: number): Promise<void> {
    await this.db.delete(fundsTable).where(eq(fundsTable.id, id));
  }

  // Custom queries
  async findByOrganization(organizationId: number): Promise<FundEntity[]> {
    const funds = await this.db
      .select()
      .from(fundsTable)
      .where(eq(fundsTable.organizationId, organizationId));

    return funds.map((f) => this.mapToEntity(f));
  }

  private mapToEntity(row: FundTableRow): FundEntity {
    return {
      ...row,
      fees: row.fees ? JSON.parse(row.fees) : null,
    };
  }
}

Standard Operations

Every repository should implement these standard operations where applicable:

CRUD Operations

interface StandardRepository<T, CreateT, UpdateT> {
  // Create
  create(data: CreateT): Promise<T>;

  // Read
  findById(id: string | number): Promise<T | null>;
  findAll(): Promise<T[]>;

  // Update
  update(id: string | number, data: UpdateT): Promise<T>;

  // Delete
  delete(id: string | number): Promise<void>;
}

Domain-Specific Queries

// Fund-specific queries
class FundRepository {
  async findByCode(code: string): Promise<FundEntity | null> {...}
  async findByOrganization(orgId: number): Promise<FundEntity[]> {...}
  async getEODState(fundId: number): Promise<EODState> {...}
}

// Investor-specific queries
class InvestorRepository {
  async getInvestorHoldings(investorId: string): Promise<InvestorHolding[]> {...}
  async getHoldingByInvestorAndFund(investorId: string, fundId: number): Promise<InvestorHolding | null> {...}
}

// Accounting-specific queries
class AccountingRepository {
  async getAccountBalance(accountId: string): Promise<AccountBalance> {...}
  }

Cloudflare D1 Considerations

Cloudflare D1 is a serverless SQLite database with specific limitations:

No Transaction Support

// ❌ WRONG: Transactions not supported
await this.db.transaction(async (tx) => {
  await tx.insert(fundsTable).values({...});
  await tx.insert(bankAccountsTable).values({...});
});

// ✅ CORRECT: Use sequential operations
async createFundWithAccount(fundData, accountData) {
  // First operation
  const fund = await this.db.insert(fundsTable).values(fundData).returning();

  // Second operation (if first succeeds)
  const account = await this.db.insert(bankAccountsTable).values({
    ...accountData,
    fundId: fund[0].id,
  }).returning();

  return { fund: fund[0], account: account[0] };
}

Statement Length Limits

// ❌ WRONG: Very long SQL with many parameters
const values = Array(1000).fill({name: 'test', value: 123});
await this.db.insert(table).values(values); // May fail

// ✅ CORRECT: Break into smaller batches
async insertMany<T>(records: T[]): Promise<void> {
  const BATCH_SIZE = 100;

  for (let i = 0; i < records.length; i += BATCH_SIZE) {
    const batch = records.slice(i, i + BATCH_SIZE);
    await this.db.insert(table).values(batch);
  }
}

Async Operations Only

// All database operations must be async
const fund = await this.db.select().from(fundsTable); // ✅
const fund = this.db.select().from(fundsTable); // ❌ Missing await

Drizzle ORM Usage

We use Drizzle ORM for type-safe database access:

Schema Definition

// worker/db/schema/funds.ts
import { sqliteTable, integer, text } from "drizzle-orm/sqlite-core";

export const fundsTable = sqliteTable("funds", {
  id: integer("id").primaryKey({ autoIncrement: true }),
  name: text("name").notNull(),
  code: text("code").notNull().unique(),
  organizationId: integer("organization_id").notNull(),
  startDate: integer("start_date", { mode: "timestamp" }).notNull(),
  fees: text("fees"), // JSON string
  eodDate: integer("eod_date", { mode: "timestamp" }),
  createdAt: integer("created_at", { mode: "timestamp" }).default(
    sql`CURRENT_TIMESTAMP`,
  ),
  updatedAt: integer("updated_at", { mode: "timestamp" }),
});

Type-Safe Queries

import { eq, and, gte, lte } from "drizzle-orm";
import { fundsTable } from "@worker/db/schema/funds";

// Select with conditions
const funds = await this.db
  .select()
  .from(fundsTable)
  .where(
    and(
      eq(fundsTable.organizationId, orgId),
      gte(fundsTable.startDate, startDate),
    ),
  );

// Join queries
const fundsWithOrg = await this.db
  .select({
    fund: fundsTable,
    org: organizationsTable,
  })
  .from(fundsTable)
  .leftJoin(
    organizationsTable,
    eq(fundsTable.organizationId, organizationsTable.id),
  );

// Aggregations
const result = await this.db
  .select({
    fundId: fundsTable.id,
    count: sql<number>`count(*)`,
  })
  .from(fundsTable)
  .groupBy(fundsTable.organizationId);

Best Practices

DO ✅

Keep repositories focused on data access
Map database types to domain entities
Use parameterized queries to prevent SQL injection
Handle null cases explicitly
Return domain entities, not raw database rows
Use Drizzle ORM for all queries
Batch large operations to respect D1 limits

DON'T ❌

Don't put business logic in repositories
Don't use raw SQL strings when Drizzle can handle it
Don't assume transactions are available
Don't ignore error handling
Don't return database-specific types to services
Don't create dependencies between repositories
Don't make HTTP calls or external API calls

Error Handling

class FundRepository {
  async findById(id: number): Promise<FundEntity | null> {
    try {
      const [fund] = await this.db
        .select()
        .from(fundsTable)
        .where(eq(fundsTable.id, id));

      return fund ? this.mapToEntity(fund) : null;
    } catch (error) {
      // Log error but let it propagate
      console.error(`Failed to fetch fund ${id}:`, error);
      throw new Error(`Database error fetching fund: ${error.message}`);
    }
  }

  async create(data: CreateFundRequest): Promise<FundEntity> {
    try {
      const [fund] = await this.db
        .insert(fundsTable)
        .values({...})
        .returning();

      return this.mapToEntity(fund);
    } catch (error) {
      // Handle unique constraint violations
      if (error.message.includes('UNIQUE constraint')) {
        throw new Error(`Fund with code ${data.code} already exists`);
      }
      throw error;
    }
  }
}

JSON Handling

Cloudflare D1 (SQLite) stores JSON as text. Repositories handle serialization:

class FundRepository {
  async create(data: CreateFundRequest): Promise<FundEntity> {
    const [fund] = await this.db
      .insert(fundsTable)
      .values({
        ...data,
        // Serialize JSON to string
        fees: data.fees ? JSON.stringify(data.fees) : null,
      })
      .returning();

    return this.mapToEntity(fund);
  }

  private mapToEntity(row: FundTableRow): FundEntity {
    return {
      ...row,
      // Deserialize JSON from string
      fees: row.fees ? JSON.parse(row.fees) : null,
    };
  }
}

Date Handling

Drizzle stores dates as Unix timestamps:

// Schema definition
export const fundsTable = sqliteTable("funds", {
  startDate: integer("start_date", { mode: "timestamp" }).notNull(),
  createdAt: integer("created_at", { mode: "timestamp" }).default(
    sql`CURRENT_TIMESTAMP`,
  ),
});

// Repository automatically converts
const fund = await this.db
  .insert(fundsTable)
  .values({
    startDate: new Date("2024-01-01"), // Automatically converted to timestamp
  })
  .returning();

// Querying by date
const funds = await this.db
  .select()
  .from(fundsTable)
  .where(gte(fundsTable.startDate, new Date("2024-01-01")));

Complex Queries Example

class PortfolioRepository {
  /**
   * Get equity holdings with market values for a specific date
   */
  async getHoldingsWithMarketData(
    fundId: number,
    asOf: Date,
  ): Promise<EquityHoldingWithPrice[]> {
    const results = await this.db
      .select({
        holding: equityHoldingsTable,
        price: marketPricesTable.closePrice,
        equity: equitiesTable,
      })
      .from(equityHoldingsTable)
      .innerJoin(
        equitiesTable,
        eq(equityHoldingsTable.equityId, equitiesTable.id),
      )
      .leftJoin(
        marketPricesTable,
        and(
          eq(marketPricesTable.securityId, equitiesTable.id),
          eq(marketPricesTable.priceDate, asOf),
        ),
      )
      .where(
        and(
          eq(equityHoldingsTable.fundId, fundId),
          lte(equityHoldingsTable.snapshotDate, asOf),
        ),
      )
      .orderBy(equityHoldingsTable.snapshotDate);

    return results.map((r) => ({
      ...r.holding,
      equity: r.equity,
      currentPrice: r.price,
      marketValue: r.holding.quantity * (r.price || 0),
    }));
  }
}

Testing Repositories

import { describe, it, expect, beforeAll } from "vitest";
import { getTestDB } from "../helpers/setup";
import { FundRepository } from "@worker/services/fund/repository";

describe("FundRepository", () => {
  let db: ProductionDBClient;
  let repository: FundRepository;

  beforeAll(async () => {
    db = await getTestDB();
    repository = new FundRepository(db);
  });

  it("should create and retrieve a fund", async () => {
    const created = await repository.create({
      name: "Test Fund",
      code: "TEST01",
      organizationId: "org-1",
      startDate: new Date(),
    });

    expect(created.id).toBeDefined();

    const retrieved = await repository.findById(created.id);
    expect(retrieved).toEqual(created);
  });

  it("should return null for non-existent fund", async () => {
    const fund = await repository.findById(99999);
    expect(fund).toBeNull();
  });
});

Purpose​

Standard Repository Structure​

Standard Operations​

CRUD Operations​

Domain-Specific Queries​

Cloudflare D1 Considerations​

No Transaction Support​

Statement Length Limits​

Async Operations Only​

Drizzle ORM Usage​

Schema Definition​

Type-Safe Queries​

Best Practices​

DO ✅​

DON'T ❌​

Error Handling​

JSON Handling​

Date Handling​

Complex Queries Example​

Testing Repositories​

See Also​