TDD Saved Me From a Database I Didn't Have Yet

The Hook

I used to think TDD was mostly ceremony for code that's obviously going to work anyway. Building the risk-scoring piece of Web3 Shield this week talked me out of that, and I want to show the actual code that did it — not just claim it happened.

The Rule Worth Isolating

The heuristic is small on purpose: look at a transaction's function signature, decide how risky it is.

def calculate_risk_score(signature: str) -> int:
    # 0x095ea7b3 = 'approve' — lets another contract spend your tokens.
    # Very high risk, heavily used in phishing and wallet drains.
    if signature == "0x095ea7b3":
        return 90

    # 0xa9059cbb = 'transfer' — standard ERC-20 transfer.
    # Moderate risk, common transaction.
    elif signature == "0xa9059cbb":
        return 30

    # Anything else, unknown contract, or native transfer.
    return 10

And the tests that came first:

def test_calculate_risk_score_approve_perigoso():
    score = calculate_risk_score("0x095ea7b3")
    assert score == 90

def test_calculate_risk_score_transferencia_comum():
    score = calculate_risk_score("0xa9059cbb")
    assert score == 30

def test_calculate_risk_score_desconhecido():
    score = calculate_risk_score("0x00000000")
    assert score == 10

Nothing here touches a database, a network call, or Docker. That's the part that clicked for me: I could run and trust these tests before Postgres, before the polling loop, before any of the plumbing existed. The business rule is a pure function, and testing it in isolation is what "the domain doesn't know about infrastructure" actually feels like, not just a phrase from a Clean Architecture talk.

The Plumbing Around It

Once the rule was trustworthy, wiring it up was almost anticlimactic — which, I'm starting to think, is the point:

def main():
    repo = PostgresRepository()
    while True:
        unprocessed = repo.get_unprocessed_transactions()
        for tx in unprocessed:
            score = calculate_risk_score(tx['function_signature'])
            repo.save_risk_analysis(tx['tx_hash'], score)
        time.sleep(3)

def get_unprocessed_transactions(self):
    query = """
        SELECT r.tx_hash, r.function_signature 
        FROM raw_transactions r
        LEFT JOIN risk_analysis a ON r.tx_hash = a.tx_hash
        WHERE a.tx_hash IS NULL;
    """
    with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(query)
        return cursor.fetchall()

A LEFT JOIN ... WHERE a.tx_hash IS NULL is the whole "find what I haven't processed yet" logic — no separate status column to keep in sync, no queue system. Postgres already knows what's missing.

Prisma Introspecting a Database It Didn't Create

The Node API layer never wrote its own schema. Go and an init.sql file had already created the tables, so I ran:

npx prisma db pull

and Prisma read the existing Postgres schema and generated matching TypeScript models automatically:

model raw_transactions {
  id                 Int       @id @default(autoincrement())
  tx_hash            String    @unique @db.VarChar(66)
  to_address         String?   @db.VarChar(42)
  function_signature String?   @db.VarChar(10)
  risk_analysis      risk_analysis?
}

model risk_analysis {
  id               Int       @id @default(autoincrement())
  tx_hash          String    @unique @db.VarChar(66)
  risk_score       Int
  raw_transactions raw_transactions @relation(fields: [tx_hash], references: [tx_hash])
}

That relation is what makes the wallet audit endpoint a single query instead of two round trips and manual stitching:

fastify.get('/wallet/:address/risk', async (request, reply) => {
  const { address } = request.params as { address: string };

  const interactions = await prisma.raw_transactions.findMany({
    where: { to_address: { equals: address, mode: 'insensitive' } },
    include: { risk_analysis: true },
  });

  const hasCriticalRisk = interactions.some(
    (tx) => tx.risk_analysis && tx.risk_analysis.risk_score >= 80
  );

  return reply.send({
    success: true,
    wallet: address,
    status: hasCriticalRisk ? 'CRITICAL_RISK' : 'MODERATE_RISK',
    total_interactions: interactions.length,
  });
});

include: { risk_analysis: true } is doing a JOIN I didn't have to write by hand.

Two Bugs That Cost Real Time

The CLI that wasn't the CLI. npx prisma started failing with confusing errors — it turned out an experimental package, @prisma/composer, had intercepted the global prisma command. Fixed by installing Prisma as a local dev dependency (npm install prisma@latest --save-dev) and calling it explicitly from package.json scripts instead of trusting global resolution.

Secrets in the wrong place. Early on, database credentials were hardcoded directly in the Go source. I moved them into .env files loaded via godotenv, and turned the schema itself into a versioned init.sql, mounted into the Postgres container through Docker Compose — so the database structure is something reviewable in a PR, not something typed once into a terminal and forgotten.

The Gap I'm Leaving In on Purpose

Right now, raw_transactions only stores to_address — the contract being called. That's enough to flag a dangerous contract, but not enough to say a specific wallet is at risk, since it never records who's calling. I know this is a gap. I'm writing it down instead of quietly fixing it off-screen, because the next real decision — adding from_address to the Go layer vs. building webhooks first — is one I haven't made yet.

Let's Discuss! 👇

If you've had a CLI tool silently hijacked by another package before (like the @prisma/composer conflict here) — what was the debugging path that actually got you to the real cause?

python #typescript #tdd #postgresql