Add Prime Intellect client, API, and MCP server
packages/prime — a hand-written typed client, because the first-party SDK is Python only. Deliberately narrow: PIG reads availability and nothing else, and the key it holds should be scoped so it could not provision even if the code tried. Rate limits are undocumented upstream, so it backs off empirically with full jitter and honours Retry-After. Unknown fields survive in `raw` rather than being dropped. apps/api — Hono, with authentication and authorization kept firmly apart. A verified JWT proves someone has an account in the identity project, which may be shared with other applications; it does NOT prove they belong here. Access requires a row in PIG's own users table, and a token without one gets 403 needs_profile rather than entry. The capacity service is the business logic: availability counts sold and held separately, so a live hold removes inventory from everyone else's availability without inflating utilisation. Expired holds are ignored at read time, so the numbers stay right even when the sweeper is behind. Matching treats interconnect as a hard filter and excludes Unknown as well as Ethernet — unverified is not the same as adequate. apps/mcp — nine tools over stdio, so a team member drives PIG from Claude Code, Codex, prime-agent, or a Buzz agent. It holds an API key and calls the same HTTP API the browser does, with no database credentials, so an agent can never reach further than the person it acts for. Results are formatted as prose rather than raw JSON. Theme preferences live in the database rather than localStorage, so a chosen accent follows someone from laptop to phone. Status colours stay independent of the accent: if "at risk" re-tinted to whatever a user picked, the signal would be gone. Note on the SDK import: its package exports use a `./*` wildcard whose types entry resolves server/mcp.js to server/mcp.js.d.ts, which does not exist. The runtime specifier must keep the .js suffix, so the types are mapped via tsconfig paths rather than by writing an import that would fail at runtime. Verified: all five packages typecheck; the MCP server constructs and registers its tools. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,535 @@
|
||||
/**
|
||||
* The HTTP application.
|
||||
*
|
||||
* Deliberately thin. Following the rule stated in the README — *intelligence
|
||||
* never lives in the API* — these handlers validate input, check authorization,
|
||||
* call a service, and serialise the result. Research, enrichment, scoring and
|
||||
* matching heuristics live in the service layer or in the agent, never here.
|
||||
*/
|
||||
import { Hono } from 'hono';
|
||||
import { cors } from 'hono/cors';
|
||||
import { logger } from 'hono/logger';
|
||||
import { and, desc, eq, ilike, isNull, or, sql } from 'drizzle-orm';
|
||||
import { z } from 'zod';
|
||||
import type { Database } from '@pig/db';
|
||||
import {
|
||||
accounts,
|
||||
activities,
|
||||
agentTasks,
|
||||
allocations,
|
||||
capacityCommitments,
|
||||
contacts,
|
||||
contracts,
|
||||
demandDeals,
|
||||
supplyDeals,
|
||||
teamMemberships,
|
||||
users,
|
||||
} from '@pig/db';
|
||||
import {
|
||||
ACCENTS,
|
||||
ACCOUNT_SIDES,
|
||||
ACTIVITY_TYPES,
|
||||
CUSTOMER_SEGMENTS,
|
||||
DEMAND_STAGES,
|
||||
SUPPLIER_TYPES,
|
||||
SUPPLY_STAGES,
|
||||
TEAMS,
|
||||
THEME_MODES,
|
||||
isValidAccent,
|
||||
isValidThemeMode,
|
||||
} from '@pig/core';
|
||||
import type { Config } from './lib/config';
|
||||
import { AuthError, createAuthenticator, type Principal } from './lib/auth';
|
||||
import { CapacityService } from './services/capacity';
|
||||
|
||||
type Env = { Variables: { principal: Principal } };
|
||||
|
||||
export function createApp(config: Config, db: Database) {
|
||||
const app = new Hono<Env>();
|
||||
const auth = createAuthenticator(config, db);
|
||||
const capacity = new CapacityService(db);
|
||||
|
||||
if (!config.isProduction) app.use('*', logger());
|
||||
|
||||
app.use(
|
||||
'/api/*',
|
||||
cors({
|
||||
// In production the front end is served from the same origin, so no
|
||||
// cross-origin allowance is needed. In development Vite runs separately.
|
||||
origin: config.isProduction ? config.PIG_PUBLIC_URL : ['http://localhost:5173'],
|
||||
credentials: true,
|
||||
}),
|
||||
);
|
||||
|
||||
/** Liveness. Unauthenticated by design so a load balancer can reach it. */
|
||||
app.get('/api/health', (c) => c.json({ ok: true, service: 'pig', version: '0.1.0' }));
|
||||
|
||||
/**
|
||||
* Public configuration for the front end — what it needs before anyone has
|
||||
* logged in. Contains only values that are safe in a browser: the anon key is
|
||||
* designed to be public, and no secret is exposed here.
|
||||
*/
|
||||
app.get('/api/config', (c) =>
|
||||
c.json({
|
||||
supabaseUrl: config.SUPABASE_URL ?? null,
|
||||
supabaseAnonKey: config.SUPABASE_ANON_KEY ?? null,
|
||||
authDisabled: !config.SUPABASE_URL,
|
||||
inviteRequired: Boolean(config.PIG_INVITE_CODE),
|
||||
accents: ACCENTS.map((a) => ({ key: a.key, label: a.label })),
|
||||
teams: TEAMS,
|
||||
}),
|
||||
);
|
||||
|
||||
// Everything below requires a principal.
|
||||
app.use('/api/*', async (c, next) => {
|
||||
const path = new URL(c.req.url).pathname;
|
||||
if (path === '/api/health' || path === '/api/config' || path === '/api/signup') {
|
||||
return next();
|
||||
}
|
||||
try {
|
||||
c.set('principal', await auth.authenticate(c.req.header('authorization')));
|
||||
} catch (error) {
|
||||
if (error instanceof AuthError) {
|
||||
return c.json({ error: error.message, code: error.code }, error.status);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
return next();
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------- identity
|
||||
|
||||
app.get('/api/me', (c) => {
|
||||
const p = c.get('principal');
|
||||
return c.json({
|
||||
id: p.userId,
|
||||
email: p.email,
|
||||
name: p.name,
|
||||
isPlatformAdmin: p.isPlatformAdmin,
|
||||
teams: p.teams,
|
||||
via: p.via,
|
||||
});
|
||||
});
|
||||
|
||||
const preferencesSchema = z.object({
|
||||
themeMode: z.enum(THEME_MODES).optional(),
|
||||
accentColor: z.string().refine(isValidAccent, 'Unknown accent').optional(),
|
||||
name: z.string().min(1).max(120).optional(),
|
||||
handle: z.string().min(2).max(40).regex(/^[a-z0-9_-]+$/i).optional(),
|
||||
title: z.string().max(160).optional(),
|
||||
timezone: z.string().max(80).optional(),
|
||||
});
|
||||
|
||||
/**
|
||||
* Appearance and profile preferences.
|
||||
*
|
||||
* Persisted server-side rather than in localStorage so that a person's chosen
|
||||
* theme follows them from laptop to phone — which matters more than it might
|
||||
* seem, because a CRM is genuinely used on both.
|
||||
*/
|
||||
app.patch('/api/me/preferences', async (c) => {
|
||||
const p = c.get('principal');
|
||||
const parsed = preferencesSchema.safeParse(await c.req.json());
|
||||
if (!parsed.success) {
|
||||
return c.json({ error: 'Invalid preferences', issues: parsed.error.issues }, 400);
|
||||
}
|
||||
const [updated] = await db
|
||||
.update(users)
|
||||
.set({ ...parsed.data, updatedAt: new Date() })
|
||||
.where(eq(users.id, p.userId))
|
||||
.returning();
|
||||
return c.json(updated);
|
||||
});
|
||||
|
||||
app.get('/api/me/profile', async (c) => {
|
||||
const p = c.get('principal');
|
||||
const [user] = await db.select().from(users).where(eq(users.id, p.userId)).limit(1);
|
||||
return c.json(user ?? null);
|
||||
});
|
||||
|
||||
app.get('/api/team', async (c) => {
|
||||
const rows = await db
|
||||
.select({
|
||||
id: users.id,
|
||||
name: users.name,
|
||||
email: users.email,
|
||||
handle: users.handle,
|
||||
title: users.title,
|
||||
avatarUrl: users.avatarUrl,
|
||||
team: teamMemberships.team,
|
||||
role: teamMemberships.role,
|
||||
})
|
||||
.from(users)
|
||||
.leftJoin(teamMemberships, eq(teamMemberships.userId, users.id))
|
||||
.where(isNull(users.deactivatedAt));
|
||||
|
||||
// Collapse the join: one row per person, carrying every team they're on.
|
||||
const byId = new Map<string, Record<string, unknown>>();
|
||||
for (const row of rows) {
|
||||
const existing = byId.get(row.id);
|
||||
const membership = row.team ? { team: row.team, role: row.role } : null;
|
||||
if (existing) {
|
||||
if (membership) (existing.teams as unknown[]).push(membership);
|
||||
} else {
|
||||
byId.set(row.id, {
|
||||
id: row.id,
|
||||
name: row.name,
|
||||
email: row.email,
|
||||
handle: row.handle,
|
||||
title: row.title,
|
||||
avatarUrl: row.avatarUrl,
|
||||
teams: membership ? [membership] : [],
|
||||
});
|
||||
}
|
||||
}
|
||||
return c.json([...byId.values()]);
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------- accounts
|
||||
|
||||
app.get('/api/accounts', async (c) => {
|
||||
const side = c.req.query('side');
|
||||
const q = c.req.query('q');
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(accounts)
|
||||
.where(
|
||||
and(
|
||||
isNull(accounts.archivedAt),
|
||||
side && side !== 'all'
|
||||
? or(eq(accounts.side, side as 'supply'), eq(accounts.side, 'both'))
|
||||
: undefined,
|
||||
q ? ilike(accounts.name, `%${q}%`) : undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(desc(accounts.lastActivityAt), accounts.name)
|
||||
.limit(200);
|
||||
return c.json(rows);
|
||||
});
|
||||
|
||||
app.get('/api/accounts/:id', async (c) => {
|
||||
const id = c.req.param('id');
|
||||
const [account] = await db.select().from(accounts).where(eq(accounts.id, id)).limit(1);
|
||||
if (!account) return c.json({ error: 'Not found' }, 404);
|
||||
|
||||
const [accountContacts, demand, supply, paperwork, recentActivity] = await Promise.all([
|
||||
db.select().from(contacts).where(eq(contacts.accountId, id)),
|
||||
db.select().from(demandDeals).where(eq(demandDeals.accountId, id)),
|
||||
db.select().from(supplyDeals).where(eq(supplyDeals.accountId, id)),
|
||||
db.select().from(contracts).where(eq(contracts.accountId, id)),
|
||||
db
|
||||
.select()
|
||||
.from(activities)
|
||||
.where(eq(activities.accountId, id))
|
||||
.orderBy(desc(activities.occurredAt))
|
||||
.limit(50),
|
||||
]);
|
||||
|
||||
return c.json({
|
||||
account,
|
||||
contacts: accountContacts,
|
||||
demandDeals: demand,
|
||||
supplyDeals: supply,
|
||||
contracts: paperwork,
|
||||
activities: recentActivity,
|
||||
});
|
||||
});
|
||||
|
||||
// Enum members are drawn from the ontology rather than retyped, so a value
|
||||
// added there is accepted here without a second edit — and, more importantly,
|
||||
// a value removed there stops validating rather than silently persisting.
|
||||
const accountSchema = z.object({
|
||||
name: z.string().min(1).max(200),
|
||||
domain: z.string().max(200).optional(),
|
||||
website: z.string().max(400).optional(),
|
||||
description: z.string().max(4000).optional(),
|
||||
side: z.enum(ACCOUNT_SIDES).default('demand'),
|
||||
supplierType: z.enum(SUPPLIER_TYPES).optional(),
|
||||
customerSegment: z.enum(CUSTOMER_SEGMENTS).optional(),
|
||||
country: z.string().max(100).optional(),
|
||||
region: z.string().max(100).optional(),
|
||||
ultimateParentName: z.string().max(200).optional(),
|
||||
ultimateParentCountry: z.string().max(100).optional(),
|
||||
});
|
||||
|
||||
app.post('/api/accounts', async (c) => {
|
||||
const p = c.get('principal');
|
||||
const parsed = accountSchema.safeParse(await c.req.json());
|
||||
if (!parsed.success) {
|
||||
return c.json({ error: 'Invalid account', issues: parsed.error.issues }, 400);
|
||||
}
|
||||
const [created] = await db
|
||||
.insert(accounts)
|
||||
.values({
|
||||
...parsed.data,
|
||||
ownerUserId: p.userId,
|
||||
source: 'manual',
|
||||
})
|
||||
.returning();
|
||||
|
||||
// Queue enrichment rather than performing it inline. The API must never
|
||||
// block on a model, and the queue survives the agent being down.
|
||||
if (created) {
|
||||
await db
|
||||
.insert(agentTasks)
|
||||
.values({
|
||||
kind: 'enrich_account',
|
||||
subject: created.id,
|
||||
reason: `New account created by ${p.name}`,
|
||||
requestedByUserId: p.userId,
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
}
|
||||
return c.json(created, 201);
|
||||
});
|
||||
|
||||
// ------------------------------------------------------------------- deals
|
||||
|
||||
app.get('/api/deals/demand', async (c) => {
|
||||
const rows = await db
|
||||
.select({
|
||||
deal: demandDeals,
|
||||
accountName: accounts.name,
|
||||
accountDomain: accounts.domain,
|
||||
})
|
||||
.from(demandDeals)
|
||||
.leftJoin(accounts, eq(accounts.id, demandDeals.accountId))
|
||||
.orderBy(desc(demandDeals.updatedAt))
|
||||
.limit(300);
|
||||
return c.json({ stages: DEMAND_STAGES, deals: rows });
|
||||
});
|
||||
|
||||
app.get('/api/deals/supply', async (c) => {
|
||||
const rows = await db
|
||||
.select({
|
||||
deal: supplyDeals,
|
||||
accountName: accounts.name,
|
||||
accountDomain: accounts.domain,
|
||||
})
|
||||
.from(supplyDeals)
|
||||
.leftJoin(accounts, eq(accounts.id, supplyDeals.accountId))
|
||||
.orderBy(desc(supplyDeals.updatedAt))
|
||||
.limit(300);
|
||||
return c.json({ stages: SUPPLY_STAGES, deals: rows });
|
||||
});
|
||||
|
||||
const stageSchema = z.object({ stage: z.string() });
|
||||
|
||||
/**
|
||||
* Stage transitions are recorded as activities, not merely written.
|
||||
*
|
||||
* "When did this move to procurement, and who moved it?" is the question
|
||||
* asked in every pipeline review, and a column that only holds the current
|
||||
* value cannot answer it.
|
||||
*/
|
||||
app.patch('/api/deals/demand/:id/stage', async (c) => {
|
||||
const p = c.get('principal');
|
||||
const id = c.req.param('id');
|
||||
const parsed = stageSchema.safeParse(await c.req.json());
|
||||
if (!parsed.success || !(DEMAND_STAGES as readonly string[]).includes(parsed.data.stage)) {
|
||||
return c.json({ error: 'Unknown stage' }, 400);
|
||||
}
|
||||
|
||||
const [before] = await db.select().from(demandDeals).where(eq(demandDeals.id, id)).limit(1);
|
||||
if (!before) return c.json({ error: 'Not found' }, 404);
|
||||
|
||||
const now = new Date();
|
||||
const [updated] = await db
|
||||
.update(demandDeals)
|
||||
.set({
|
||||
stage: parsed.data.stage as 'qualification',
|
||||
stageChangedAt: now,
|
||||
updatedAt: now,
|
||||
lastActivityAt: now,
|
||||
})
|
||||
.where(eq(demandDeals.id, id))
|
||||
.returning();
|
||||
|
||||
await db.insert(activities).values({
|
||||
type: 'stage_change',
|
||||
subject: `${before.stage} → ${parsed.data.stage}`,
|
||||
accountId: before.accountId,
|
||||
demandDealId: id,
|
||||
actorUserId: p.userId,
|
||||
occurredAt: now,
|
||||
meta: { from: before.stage, to: parsed.data.stage },
|
||||
});
|
||||
|
||||
return c.json(updated);
|
||||
});
|
||||
|
||||
// ------------------------------------------------------------- activities
|
||||
|
||||
const activitySchema = z.object({
|
||||
accountId: z.string().uuid().optional(),
|
||||
contactId: z.string().uuid().optional(),
|
||||
demandDealId: z.string().uuid().optional(),
|
||||
supplyDealId: z.string().uuid().optional(),
|
||||
type: z.enum(ACTIVITY_TYPES),
|
||||
subject: z.string().min(1).max(200),
|
||||
body: z.string().max(8000).optional(),
|
||||
occurredAt: z.string().datetime().optional(),
|
||||
externalId: z.string().max(200).optional(),
|
||||
});
|
||||
|
||||
app.post('/api/activities', async (c) => {
|
||||
const p = c.get('principal');
|
||||
const parsed = activitySchema.safeParse(await c.req.json());
|
||||
if (!parsed.success) {
|
||||
return c.json({ error: 'Invalid activity', issues: parsed.error.issues }, 400);
|
||||
}
|
||||
const { occurredAt, ...rest } = parsed.data;
|
||||
const when = occurredAt ? new Date(occurredAt) : new Date();
|
||||
|
||||
const [created] = await db
|
||||
.insert(activities)
|
||||
.values({
|
||||
...rest,
|
||||
occurredAt: when,
|
||||
actorUserId: p.userId,
|
||||
// An agent acting for someone is recorded as such, so the log
|
||||
// distinguishes what a person did from what was done on their behalf.
|
||||
actorAgent: p.via === 'api_key' ? 'agent' : null,
|
||||
source: p.via === 'api_key' ? 'agent' : 'manual',
|
||||
})
|
||||
// An `externalId` collision means this event was already synced from
|
||||
// Slack or Buzz; silently ignoring the duplicate keeps sync idempotent.
|
||||
.onConflictDoNothing()
|
||||
.returning();
|
||||
|
||||
if (rest.accountId) {
|
||||
await db
|
||||
.update(accounts)
|
||||
.set({ lastActivityAt: when })
|
||||
.where(eq(accounts.id, rest.accountId));
|
||||
}
|
||||
|
||||
return c.json(created ?? { deduplicated: true }, created ? 201 : 200);
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------- capacity
|
||||
|
||||
app.get('/api/capacity/availability', async (c) => {
|
||||
const gpuType = c.req.query('gpuType') ?? undefined;
|
||||
return c.json(await capacity.availability({ gpuType }));
|
||||
});
|
||||
|
||||
app.get('/api/capacity/idle', async (c) => {
|
||||
const threshold = Number(c.req.query('threshold') ?? '0.25');
|
||||
return c.json(
|
||||
await capacity.idleCapacity({
|
||||
thresholdPct: Number.isFinite(threshold) ? threshold : 0.25,
|
||||
withinDays: Number(c.req.query('withinDays') ?? '30') || 30,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
app.get('/api/capacity/margin', async (c) => c.json(await capacity.marginReport()));
|
||||
|
||||
const matchSchema = z.object({
|
||||
gpuType: z.string().optional(),
|
||||
gpuTypeAlternatives: z.array(z.string()).optional(),
|
||||
gpuCount: z.number().int().positive(),
|
||||
totalGpuHours: z.number().positive().optional(),
|
||||
requiresHighSpeedInterconnect: z.boolean().optional(),
|
||||
minSecurityTier: z.enum(['secure_cloud', 'community_cloud']).optional(),
|
||||
startsAt: z.string().datetime().optional(),
|
||||
endsAt: z.string().datetime().optional(),
|
||||
maxPricePerGpuHourCents: z.number().int().positive().optional(),
|
||||
});
|
||||
|
||||
app.post('/api/capacity/match', async (c) => {
|
||||
const parsed = matchSchema.safeParse(await c.req.json());
|
||||
if (!parsed.success) {
|
||||
return c.json({ error: 'Invalid requirement', issues: parsed.error.issues }, 400);
|
||||
}
|
||||
const { startsAt, endsAt, ...rest } = parsed.data;
|
||||
return c.json(
|
||||
await capacity.match({
|
||||
...rest,
|
||||
startsAt: startsAt ? new Date(startsAt) : undefined,
|
||||
endsAt: endsAt ? new Date(endsAt) : undefined,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
app.get('/api/inventory', async (c) =>
|
||||
c.json(
|
||||
await capacity.searchInventory({
|
||||
gpuType: c.req.query('gpuType') ?? undefined,
|
||||
minGpuCount: Number(c.req.query('minGpuCount')) || undefined,
|
||||
maxPriceCents: Number(c.req.query('maxPriceCents')) || undefined,
|
||||
requiresHighSpeedInterconnect: c.req.query('fastFabric') === 'true',
|
||||
limit: Number(c.req.query('limit')) || 50,
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
app.get('/api/commitments', async (c) => {
|
||||
const rows = await db
|
||||
.select({ commitment: capacityCommitments, accountName: accounts.name })
|
||||
.from(capacityCommitments)
|
||||
.leftJoin(accounts, eq(accounts.id, capacityCommitments.accountId))
|
||||
.orderBy(desc(capacityCommitments.startsAt));
|
||||
return c.json(rows);
|
||||
});
|
||||
|
||||
app.get('/api/allocations', async (c) => {
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(allocations)
|
||||
.orderBy(desc(allocations.startsAt))
|
||||
.limit(300);
|
||||
return c.json(rows);
|
||||
});
|
||||
|
||||
// --------------------------------------------------------------- dashboard
|
||||
|
||||
/**
|
||||
* The landing view. One round trip rather than six, because this renders on
|
||||
* a phone on a cellular connection as often as on a desk.
|
||||
*/
|
||||
app.get('/api/dashboard', async (c) => {
|
||||
const p = c.get('principal');
|
||||
const [margin, idle, openDemand, openSupply, recent] = await Promise.all([
|
||||
capacity.marginReport(),
|
||||
capacity.idleCapacity({ thresholdPct: 0.2 }),
|
||||
db
|
||||
.select({ count: sql<number>`count(*)::int` })
|
||||
.from(demandDeals)
|
||||
.where(sql`${demandDeals.stage} NOT IN ('closed_won','closed_lost')`),
|
||||
db
|
||||
.select({ count: sql<number>`count(*)::int` })
|
||||
.from(supplyDeals)
|
||||
.where(sql`${supplyDeals.stage} NOT IN ('live','churned','rejected')`),
|
||||
db
|
||||
.select()
|
||||
.from(activities)
|
||||
.orderBy(desc(activities.occurredAt))
|
||||
.limit(12),
|
||||
]);
|
||||
|
||||
return c.json({
|
||||
me: { name: p.name, teams: p.teams },
|
||||
margin: margin.totals,
|
||||
blocks: margin.blocks.length,
|
||||
idleAlerts: idle.slice(0, 5),
|
||||
openDemandDeals: openDemand[0]?.count ?? 0,
|
||||
openSupplyDeals: openSupply[0]?.count ?? 0,
|
||||
recentActivity: recent,
|
||||
});
|
||||
});
|
||||
|
||||
app.onError((error, c) => {
|
||||
if (error instanceof AuthError) {
|
||||
return c.json({ error: error.message, code: error.code }, error.status);
|
||||
}
|
||||
console.error('[pig] unhandled error', error);
|
||||
// Never leak internals to a client; the detail is in the server log.
|
||||
return c.json({ error: 'Internal error' }, 500);
|
||||
});
|
||||
|
||||
app.notFound((c) => c.json({ error: 'Not found' }, 404));
|
||||
|
||||
return app;
|
||||
}
|
||||
@@ -0,0 +1,229 @@
|
||||
/**
|
||||
* Authentication and authorization.
|
||||
*
|
||||
* The distinction is the whole point of this file, and it is the thing most
|
||||
* likely to be got wrong by someone extending PIG later:
|
||||
*
|
||||
* **Authentication** answers "who is this?" and is delegated to Supabase.
|
||||
* PIG verifies the JWT against the project's JWKS. It stores no passwords
|
||||
* and issues no sessions of its own.
|
||||
*
|
||||
* **Authorization** answers "may they use PIG?" and is answered ONLY by a row
|
||||
* in PIG's `users` table.
|
||||
*
|
||||
* These must stay separate because the Supabase project may be shared with
|
||||
* other applications. A valid token proves someone has an account *somewhere in
|
||||
* that project* — not that they belong here. Treating a verified token as
|
||||
* sufficient would silently grant every user of every sibling application full
|
||||
* access to the CRM.
|
||||
*
|
||||
* A token with no matching PIG user gets 403 with `needs_profile`, which the
|
||||
* front end turns into the invite-redemption screen.
|
||||
*/
|
||||
import { createRemoteJWKSet, jwtVerify } from 'jose';
|
||||
import { eq } from 'drizzle-orm';
|
||||
import type { Database } from '@pig/db';
|
||||
import { apiKeys, teamMemberships, users } from '@pig/db';
|
||||
import type { Team, TeamRole } from '@pig/core';
|
||||
import { createHash, timingSafeEqual } from 'node:crypto';
|
||||
import type { Config } from './config';
|
||||
|
||||
export interface Principal {
|
||||
userId: string;
|
||||
email: string;
|
||||
name: string;
|
||||
isPlatformAdmin: boolean;
|
||||
teams: { team: Team; role: TeamRole }[];
|
||||
/** How this request authenticated. Agents get their own audit trail. */
|
||||
via: 'jwt' | 'api_key' | 'development';
|
||||
apiKeyId?: string;
|
||||
scopes: string[];
|
||||
}
|
||||
|
||||
export class AuthError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
readonly status: 401 | 403,
|
||||
readonly code: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = 'AuthError';
|
||||
}
|
||||
}
|
||||
|
||||
export function createAuthenticator(config: Config, db: Database) {
|
||||
// The JWKS is fetched lazily and cached by `jose`, which also handles key
|
||||
// rotation. Building it once avoids a fetch per request.
|
||||
const jwks = config.SUPABASE_URL
|
||||
? createRemoteJWKSet(new URL(`${config.SUPABASE_URL}/auth/v1/.well-known/jwks.json`))
|
||||
: null;
|
||||
|
||||
async function loadPrincipal(
|
||||
userId: string,
|
||||
via: Principal['via'],
|
||||
extras: { apiKeyId?: string; scopes?: string[] } = {},
|
||||
): Promise<Principal> {
|
||||
const [user] = await db.select().from(users).where(eq(users.id, userId)).limit(1);
|
||||
if (!user) throw new AuthError('No PIG profile for this account.', 403, 'needs_profile');
|
||||
if (user.deactivatedAt) throw new AuthError('This account is deactivated.', 403, 'deactivated');
|
||||
|
||||
const memberships = await db
|
||||
.select({ team: teamMemberships.team, role: teamMemberships.role })
|
||||
.from(teamMemberships)
|
||||
.where(eq(teamMemberships.userId, user.id));
|
||||
|
||||
return {
|
||||
userId: user.id,
|
||||
email: user.email,
|
||||
name: user.name,
|
||||
// Admin rights come from the database, but the environment allowlist can
|
||||
// grant them too — that is how the first admin exists before anyone has
|
||||
// been able to log in and promote anybody.
|
||||
isPlatformAdmin:
|
||||
user.isPlatformAdmin || config.adminEmails.includes(user.email.toLowerCase()),
|
||||
teams: memberships as { team: Team; role: TeamRole }[],
|
||||
via,
|
||||
apiKeyId: extras.apiKeyId,
|
||||
scopes: extras.scopes ?? ['read', 'write'],
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
/**
|
||||
* Resolve the principal for a request, or throw.
|
||||
*
|
||||
* Accepts either a Supabase JWT or a PIG API key, both in the
|
||||
* Authorization header. API keys exist so that an agent acting for a person
|
||||
* is a distinct principal from that person — separately revocable, with its
|
||||
* own audit trail and its own scopes.
|
||||
*/
|
||||
async authenticate(header: string | undefined): Promise<Principal> {
|
||||
// Development escape hatch. Guarded three ways, and `loadConfig` refuses
|
||||
// to start in production without Supabase, so this cannot leak into a
|
||||
// real deployment.
|
||||
if (!config.SUPABASE_URL && !config.isProduction) {
|
||||
const [devUser] = await db.select().from(users).limit(1);
|
||||
if (!devUser) {
|
||||
throw new AuthError(
|
||||
'Auth is disabled and the database has no users. Run `npm run db:seed`.',
|
||||
403,
|
||||
'no_dev_user',
|
||||
);
|
||||
}
|
||||
return loadPrincipal(devUser.id, 'development');
|
||||
}
|
||||
|
||||
if (!header?.startsWith('Bearer ')) {
|
||||
throw new AuthError('Missing bearer token.', 401, 'no_token');
|
||||
}
|
||||
const token = header.slice(7).trim();
|
||||
|
||||
// PIG-issued API keys carry a recognisable prefix, so we can route
|
||||
// without attempting an expensive and pointless JWT verification.
|
||||
if (token.startsWith('pig_')) return authenticateApiKey(token);
|
||||
|
||||
if (!jwks) throw new AuthError('Authentication is not configured.', 401, 'no_jwks');
|
||||
|
||||
let subject: string;
|
||||
try {
|
||||
const { payload } = await jwtVerify(token, jwks, {
|
||||
// Supabase signs with the project URL as issuer.
|
||||
issuer: `${config.SUPABASE_URL}/auth/v1`,
|
||||
});
|
||||
if (!payload.sub) throw new Error('token has no subject');
|
||||
subject = payload.sub;
|
||||
} catch {
|
||||
// Deliberately opaque: distinguishing "expired" from "malformed" from
|
||||
// "wrong issuer" tells an attacker which knob to turn.
|
||||
throw new AuthError('Invalid or expired token.', 401, 'invalid_token');
|
||||
}
|
||||
|
||||
const [user] = await db
|
||||
.select({ id: users.id })
|
||||
.from(users)
|
||||
.where(eq(users.authSubject, subject))
|
||||
.limit(1);
|
||||
|
||||
if (!user) {
|
||||
// Authenticated but not authorized — the case that matters when the
|
||||
// identity provider is shared with another application.
|
||||
throw new AuthError(
|
||||
'This account is not a member of this PIG workspace.',
|
||||
403,
|
||||
'needs_profile',
|
||||
);
|
||||
}
|
||||
|
||||
return loadPrincipal(user.id, 'jwt');
|
||||
},
|
||||
};
|
||||
|
||||
async function authenticateApiKey(token: string): Promise<Principal> {
|
||||
const hash = hashApiKey(token);
|
||||
const [record] = await db
|
||||
.select()
|
||||
.from(apiKeys)
|
||||
.where(eq(apiKeys.keyHash, hash))
|
||||
.limit(1);
|
||||
|
||||
if (!record) throw new AuthError('Unknown API key.', 401, 'invalid_key');
|
||||
if (record.revokedAt) throw new AuthError('This API key was revoked.', 401, 'revoked_key');
|
||||
if (record.expiresAt && record.expiresAt < new Date()) {
|
||||
throw new AuthError('This API key has expired.', 401, 'expired_key');
|
||||
}
|
||||
|
||||
// Best-effort last-used stamp. Never block the request on it: a failed
|
||||
// bookkeeping write must not deny access.
|
||||
void db
|
||||
.update(apiKeys)
|
||||
.set({ lastUsedAt: new Date() })
|
||||
.where(eq(apiKeys.id, record.id))
|
||||
.catch(() => {});
|
||||
|
||||
return loadPrincipal(record.userId, 'api_key', {
|
||||
apiKeyId: record.id,
|
||||
scopes: record.scopes,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash an API key for storage and lookup.
|
||||
*
|
||||
* SHA-256 without a salt is correct here, unlike for passwords: the key is 256
|
||||
* bits of machine-generated randomness, so there is no dictionary to attack and
|
||||
* lookup must be deterministic. What matters is that the plaintext is never
|
||||
* stored, so a database leak yields no working credentials.
|
||||
*/
|
||||
export function hashApiKey(key: string): string {
|
||||
return createHash('sha256').update(key).digest('hex');
|
||||
}
|
||||
|
||||
/** Constant-time compare, for anywhere a secret is checked directly. */
|
||||
export function safeEqual(a: string, b: string): boolean {
|
||||
const ab = Buffer.from(a);
|
||||
const bb = Buffer.from(b);
|
||||
// Length alone can leak, so compare hashes of equal length rather than
|
||||
// returning early on a length mismatch.
|
||||
const ah = createHash('sha256').update(ab).digest();
|
||||
const bh = createHash('sha256').update(bb).digest();
|
||||
return timingSafeEqual(ah, bh);
|
||||
}
|
||||
|
||||
/** Does this principal belong to the team, at or above the given role? */
|
||||
export function hasTeamAccess(
|
||||
principal: Principal,
|
||||
team: Team,
|
||||
minimumRole: TeamRole = 'member',
|
||||
): boolean {
|
||||
if (principal.isPlatformAdmin) return true;
|
||||
const membership = principal.teams.find((t) => t.team === team);
|
||||
if (!membership) return false;
|
||||
const rank: Record<TeamRole, number> = { member: 0, lead: 1, admin: 2 };
|
||||
return rank[membership.role] >= rank[minimumRole];
|
||||
}
|
||||
|
||||
export function requireScope(principal: Principal, scope: string): void {
|
||||
if (principal.scopes.includes(scope)) return;
|
||||
throw new AuthError(`This credential lacks the '${scope}' scope.`, 403, 'insufficient_scope');
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
/**
|
||||
* Configuration, read once at boot and validated loudly.
|
||||
*
|
||||
* A misconfigured deployment should fail to start with a clear message rather
|
||||
* than start successfully and behave subtly wrong. The warnings below are the
|
||||
* cases where PIG *can* run but an operator almost certainly did not intend
|
||||
* the resulting behaviour.
|
||||
*/
|
||||
import { z } from 'zod';
|
||||
|
||||
const schema = z.object({
|
||||
DATABASE_URL: z.string().min(1, 'DATABASE_URL is required.'),
|
||||
|
||||
SUPABASE_URL: z.string().url().optional(),
|
||||
SUPABASE_ANON_KEY: z.string().optional(),
|
||||
SUPABASE_SERVICE_KEY: z.string().optional(),
|
||||
|
||||
PIG_PORT: z.coerce.number().int().positive().default(8920),
|
||||
PIG_PUBLIC_URL: z.string().default('http://localhost:8920'),
|
||||
NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
|
||||
|
||||
PIG_ADMIN_EMAILS: z.string().default(''),
|
||||
PIG_INVITE_CODE: z.string().optional(),
|
||||
|
||||
PRIME_API_KEY: z.string().optional(),
|
||||
PRIME_API_BASE: z.string().default('https://api.primeintellect.ai'),
|
||||
PRIME_SYNC_ENABLED: z.coerce.boolean().default(false),
|
||||
PRIME_SYNC_INTERVAL_MINUTES: z.coerce.number().int().positive().default(30),
|
||||
|
||||
PIGGY_ENABLED: z.coerce.boolean().default(false),
|
||||
ANTHROPIC_API_KEY: z.string().optional(),
|
||||
PIGGY_MODEL: z.string().default('claude-sonnet-5'),
|
||||
PIGGY_LEASE_SECONDS: z.coerce.number().int().positive().default(300),
|
||||
|
||||
SLACK_BOT_TOKEN: z.string().optional(),
|
||||
SLACK_SIGNING_SECRET: z.string().optional(),
|
||||
BUZZ_RELAY_URL: z.string().optional(),
|
||||
});
|
||||
|
||||
export type Config = z.infer<typeof schema> & {
|
||||
adminEmails: string[];
|
||||
isProduction: boolean;
|
||||
};
|
||||
|
||||
export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config {
|
||||
const parsed = schema.safeParse(env);
|
||||
if (!parsed.success) {
|
||||
const issues = parsed.error.issues.map((i) => ` ${i.path.join('.')}: ${i.message}`);
|
||||
throw new Error(`Invalid configuration:\n${issues.join('\n')}`);
|
||||
}
|
||||
|
||||
const adminEmails = parsed.data.PIG_ADMIN_EMAILS.split(',')
|
||||
.map((e) => e.trim().toLowerCase())
|
||||
.filter(Boolean);
|
||||
|
||||
const config: Config = {
|
||||
...parsed.data,
|
||||
adminEmails,
|
||||
isProduction: parsed.data.NODE_ENV === 'production',
|
||||
};
|
||||
|
||||
warnOnFootguns(config);
|
||||
return config;
|
||||
}
|
||||
|
||||
function warnOnFootguns(config: Config): void {
|
||||
const warn = (message: string) => console.warn(`[pig] WARNING: ${message}`);
|
||||
|
||||
if (config.adminEmails.length === 0) {
|
||||
// Safe default, but worth saying out loud: with no admins nobody can
|
||||
// manage invites or settings through the UI.
|
||||
warn('PIG_ADMIN_EMAILS is empty — no user will have platform-admin rights.');
|
||||
}
|
||||
|
||||
if (!config.SUPABASE_URL) {
|
||||
warn(
|
||||
'SUPABASE_URL is not set — authentication is DISABLED and every request ' +
|
||||
'runs as the development user. Never do this in production.',
|
||||
);
|
||||
}
|
||||
|
||||
if (config.isProduction && !config.SUPABASE_URL) {
|
||||
throw new Error(
|
||||
'Refusing to start: NODE_ENV=production with no SUPABASE_URL would serve ' +
|
||||
'the entire CRM unauthenticated.',
|
||||
);
|
||||
}
|
||||
|
||||
if (config.PRIME_SYNC_ENABLED && !config.PRIME_API_KEY) {
|
||||
warn('PRIME_SYNC_ENABLED is on but PRIME_API_KEY is unset — sync will not run.');
|
||||
}
|
||||
|
||||
if (config.PIGGY_ENABLED && !config.ANTHROPIC_API_KEY) {
|
||||
warn('PIGGY_ENABLED is on but ANTHROPIC_API_KEY is unset — the agent will idle.');
|
||||
}
|
||||
|
||||
if (config.SUPABASE_SERVICE_KEY) {
|
||||
// Present legitimately for admin provisioning, but it is the most powerful
|
||||
// credential in the deployment and most installs do not need it.
|
||||
warn(
|
||||
'SUPABASE_SERVICE_KEY is set. It is only needed for administrative user ' +
|
||||
'provisioning; unset it if you are not using that.',
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
/**
|
||||
* The PIG server.
|
||||
*
|
||||
* Serves the API and, in production, the built front end from the same origin.
|
||||
* Same-origin matters more than it might appear: the browser holds its auth
|
||||
* session per origin, so splitting the app across two hostnames turns sign-in
|
||||
* into a loop that looks like a broken deployment.
|
||||
*/
|
||||
import { serve } from '@hono/node-server';
|
||||
import { serveStatic } from '@hono/node-server/serve-static';
|
||||
import { existsSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { createDatabase } from '@pig/db';
|
||||
import { createApp } from './app';
|
||||
import { loadConfig } from './lib/config';
|
||||
import { startPrimeSync } from './services/sync';
|
||||
import { CapacityService } from './services/capacity';
|
||||
|
||||
const config = loadConfig();
|
||||
const db = createDatabase({ url: config.DATABASE_URL });
|
||||
const app = createApp(config, db);
|
||||
|
||||
// Serve the built SPA when it exists. Absent in development, where Vite serves
|
||||
// it on its own port with hot reload.
|
||||
const webDist = join(process.cwd(), 'apps/web/dist');
|
||||
if (existsSync(webDist)) {
|
||||
app.use('/assets/*', serveStatic({ root: './apps/web/dist' }));
|
||||
app.get('*', serveStatic({ path: './apps/web/dist/index.html' }));
|
||||
console.log('[pig] serving front end from', webDist);
|
||||
}
|
||||
|
||||
const server = serve({ fetch: app.fetch, port: config.PIG_PORT }, (info) => {
|
||||
console.log(`[pig] listening on http://localhost:${info.port}`);
|
||||
console.log(`[pig] environment: ${config.NODE_ENV}`);
|
||||
});
|
||||
|
||||
// Background work. Both are optional and the application is fully usable with
|
||||
// neither running.
|
||||
const stopSync = startPrimeSync(config, db);
|
||||
|
||||
const capacity = new CapacityService(db);
|
||||
const holdSweeper = setInterval(
|
||||
() => {
|
||||
void capacity
|
||||
.sweepExpiredHolds()
|
||||
.then((n) => n > 0 && console.log(`[pig] released ${n} expired capacity hold(s)`))
|
||||
.catch((error) => console.error('[pig] hold sweep failed', error));
|
||||
},
|
||||
5 * 60 * 1000,
|
||||
);
|
||||
|
||||
/**
|
||||
* Shut down cleanly so that in-flight requests finish and Postgres connections
|
||||
* are returned rather than left for the server to reap.
|
||||
*/
|
||||
function shutdown(signal: string) {
|
||||
console.log(`[pig] ${signal} received, shutting down`);
|
||||
clearInterval(holdSweeper);
|
||||
stopSync();
|
||||
server.close(() => process.exit(0));
|
||||
// Do not hang forever if a connection refuses to drain.
|
||||
setTimeout(() => process.exit(1), 10_000).unref();
|
||||
}
|
||||
|
||||
process.on('SIGTERM', () => shutdown('SIGTERM'));
|
||||
process.on('SIGINT', () => shutdown('SIGINT'));
|
||||
@@ -0,0 +1,418 @@
|
||||
/**
|
||||
* Capacity: availability, matching and margin.
|
||||
*
|
||||
* This is PIG's business logic. Everything here answers one of three questions
|
||||
* a compute GTM team asks constantly and cannot ask a generic CRM at all:
|
||||
*
|
||||
* "What can I actually sell?" → availability
|
||||
* "What fits this customer?" → matching
|
||||
* "What is it worth?" → margin
|
||||
*/
|
||||
import { and, eq, gte, inArray, isNull, lte, or, sql } from 'drizzle-orm';
|
||||
import type { Database } from '@pig/db';
|
||||
import {
|
||||
allocations,
|
||||
capacityCommitments,
|
||||
CONSUMING_ALLOCATION_STATUSES,
|
||||
RESERVING_ALLOCATION_STATUSES,
|
||||
inventoryListings,
|
||||
} from '@pig/db';
|
||||
import { computeMargin, breakEvenPricePerGpuHourCents } from '@pig/core';
|
||||
import type { InterconnectType, SecurityTier } from '@pig/core';
|
||||
|
||||
export interface CommitmentShape {
|
||||
intervals: string[];
|
||||
quantities: number[];
|
||||
}
|
||||
|
||||
/**
|
||||
* GPUs held at an instant, according to a commitment's shape.
|
||||
*
|
||||
* A commitment is not a rectangle: it ramps across tranches and steps down at
|
||||
* checkpoints. Where no shape is recorded the flat `gpuCount` applies for the
|
||||
* whole window, which is the common simple case.
|
||||
*/
|
||||
export function quantityAt(
|
||||
shape: CommitmentShape | null | undefined,
|
||||
flatCount: number,
|
||||
at: Date,
|
||||
): number {
|
||||
if (!shape || shape.intervals.length < 2) return flatCount;
|
||||
const t = at.getTime();
|
||||
for (let i = 0; i < shape.quantities.length; i++) {
|
||||
const start = Date.parse(shape.intervals[i]!);
|
||||
const end = Date.parse(shape.intervals[i + 1]!);
|
||||
if (Number.isFinite(start) && Number.isFinite(end) && t >= start && t < end) {
|
||||
return shape.quantities[i] ?? 0;
|
||||
}
|
||||
}
|
||||
// Outside every declared interval the commitment holds nothing. Falling back
|
||||
// to the flat count here would invent capacity beyond the contract.
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Total GPU-hours a commitment provides, integrating over its shape.
|
||||
*
|
||||
* Used to reconcile a recorded `totalGpuHours` against the shape actually
|
||||
* entered — a mismatch usually means someone typed a headline number from a
|
||||
* term sheet and then entered a ramp that does not add up to it.
|
||||
*/
|
||||
export function gpuHoursFromShape(shape: CommitmentShape): number {
|
||||
let hours = 0;
|
||||
for (let i = 0; i < shape.quantities.length; i++) {
|
||||
const start = Date.parse(shape.intervals[i]!);
|
||||
const end = Date.parse(shape.intervals[i + 1]!);
|
||||
if (!Number.isFinite(start) || !Number.isFinite(end) || end <= start) continue;
|
||||
hours += ((end - start) / 3_600_000) * (shape.quantities[i] ?? 0);
|
||||
}
|
||||
return hours;
|
||||
}
|
||||
|
||||
export interface AvailabilityRow {
|
||||
commitmentId: string;
|
||||
name: string;
|
||||
gpuType: string;
|
||||
gpuCount: number;
|
||||
interconnectType: InterconnectType;
|
||||
securityTier: SecurityTier;
|
||||
startsAt: Date;
|
||||
endsAt: Date;
|
||||
totalGpuHours: number;
|
||||
/** Hours consumed by allocations that count as sold. */
|
||||
soldGpuHours: number;
|
||||
/** Hours held by unexpired holds. Reserved, but not yet sold. */
|
||||
heldGpuHours: number;
|
||||
/** Hours neither sold nor held. What a seller may actually offer. */
|
||||
availableGpuHours: number;
|
||||
costPerGpuHourCents: number;
|
||||
utilisation: number;
|
||||
/** Price at which the remaining hours break even on this block. */
|
||||
breakEvenPriceCents: number | null;
|
||||
}
|
||||
|
||||
export class CapacityService {
|
||||
constructor(private readonly db: Database) {}
|
||||
|
||||
/**
|
||||
* What is genuinely sellable, per commitment.
|
||||
*
|
||||
* The key subtlety is that sold and held are counted separately. A live hold
|
||||
* must remove capacity from availability — otherwise two sellers promise the
|
||||
* same GPUs — but it is not revenue and must not inflate utilisation. A
|
||||
* pipeline of optimistic holds should never be able to make the book look
|
||||
* full.
|
||||
*
|
||||
* Expired holds are ignored here rather than requiring a sweep to have run,
|
||||
* so availability is correct even if the cleanup job is behind.
|
||||
*/
|
||||
async availability(options: { at?: Date; gpuType?: string } = {}): Promise<AvailabilityRow[]> {
|
||||
const now = options.at ?? new Date();
|
||||
|
||||
const commitments = await this.db
|
||||
.select()
|
||||
.from(capacityCommitments)
|
||||
.where(
|
||||
and(
|
||||
isNull(capacityCommitments.terminatedAt),
|
||||
gte(capacityCommitments.endsAt, now),
|
||||
options.gpuType ? eq(capacityCommitments.gpuType, options.gpuType) : undefined,
|
||||
),
|
||||
);
|
||||
|
||||
if (commitments.length === 0) return [];
|
||||
|
||||
const ids = commitments.map((c) => c.id);
|
||||
const allocRows = await this.db
|
||||
.select()
|
||||
.from(allocations)
|
||||
.where(
|
||||
and(
|
||||
inArray(allocations.capacityCommitmentId, ids),
|
||||
inArray(allocations.status, [...RESERVING_ALLOCATION_STATUSES]),
|
||||
),
|
||||
);
|
||||
|
||||
return commitments.map((commitment) => {
|
||||
const mine = allocRows.filter((a) => a.capacityCommitmentId === commitment.id);
|
||||
|
||||
let soldGpuHours = 0;
|
||||
let heldGpuHours = 0;
|
||||
for (const a of mine) {
|
||||
const hours = Number(a.gpuHours);
|
||||
if ((CONSUMING_ALLOCATION_STATUSES as readonly string[]).includes(a.status)) {
|
||||
soldGpuHours += hours;
|
||||
} else if (!a.holdExpiresAt || a.holdExpiresAt > now) {
|
||||
// A lapsed hold reserves nothing, whether or not it has been swept.
|
||||
heldGpuHours += hours;
|
||||
}
|
||||
}
|
||||
|
||||
const totalGpuHours = Number(commitment.totalGpuHours);
|
||||
const margin = computeMargin(
|
||||
{ gpuHours: totalGpuHours, costPerGpuHourCents: commitment.costPerGpuHourCents },
|
||||
mine
|
||||
.filter((a) => (CONSUMING_ALLOCATION_STATUSES as readonly string[]).includes(a.status))
|
||||
.map((a) => ({
|
||||
gpuHours: Number(a.gpuHours),
|
||||
pricePerGpuHourCents: a.pricePerGpuHourCents,
|
||||
})),
|
||||
);
|
||||
|
||||
return {
|
||||
commitmentId: commitment.id,
|
||||
name: commitment.name,
|
||||
gpuType: commitment.gpuType,
|
||||
gpuCount: commitment.gpuCount,
|
||||
interconnectType: commitment.interconnectType as InterconnectType,
|
||||
securityTier: commitment.securityTier as SecurityTier,
|
||||
startsAt: commitment.startsAt,
|
||||
endsAt: commitment.endsAt,
|
||||
totalGpuHours,
|
||||
soldGpuHours,
|
||||
heldGpuHours,
|
||||
availableGpuHours: Math.max(0, totalGpuHours - soldGpuHours - heldGpuHours),
|
||||
costPerGpuHourCents: commitment.costPerGpuHourCents,
|
||||
utilisation: margin.utilisation,
|
||||
breakEvenPriceCents: breakEvenPricePerGpuHourCents(
|
||||
{ gpuHours: totalGpuHours, costPerGpuHourCents: commitment.costPerGpuHourCents },
|
||||
mine
|
||||
.filter((a) =>
|
||||
(CONSUMING_ALLOCATION_STATUSES as readonly string[]).includes(a.status),
|
||||
)
|
||||
.map((a) => ({
|
||||
gpuHours: Number(a.gpuHours),
|
||||
pricePerGpuHourCents: a.pricePerGpuHourCents,
|
||||
})),
|
||||
),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Match a customer requirement against capacity we hold.
|
||||
*
|
||||
* Interconnect is a hard filter rather than a preference. Selling
|
||||
* Ethernet-only capacity to a distributed-training customer produces a
|
||||
* cluster that cannot do the job, which is worse than losing the deal — so a
|
||||
* requirement for high-speed fabric excludes `Ethernet` and `Unknown` alike.
|
||||
* `Unknown` is excluded deliberately: unverified is not the same as adequate.
|
||||
*/
|
||||
async match(requirement: {
|
||||
gpuType?: string;
|
||||
gpuTypeAlternatives?: string[];
|
||||
gpuCount: number;
|
||||
totalGpuHours?: number;
|
||||
requiresHighSpeedInterconnect?: boolean;
|
||||
minSecurityTier?: SecurityTier;
|
||||
startsAt?: Date;
|
||||
endsAt?: Date;
|
||||
maxPricePerGpuHourCents?: number;
|
||||
}): Promise<
|
||||
(AvailabilityRow & {
|
||||
/** 0–1. Higher is a better fit. */
|
||||
score: number;
|
||||
/** Why this matched, in plain words, for the UI and for agents. */
|
||||
rationale: string[];
|
||||
})[]
|
||||
> {
|
||||
const acceptableTypes = [
|
||||
...(requirement.gpuType ? [requirement.gpuType] : []),
|
||||
...(requirement.gpuTypeAlternatives ?? []),
|
||||
];
|
||||
|
||||
const rows = await this.availability({ at: requirement.startsAt ?? new Date() });
|
||||
|
||||
const matches = rows
|
||||
.filter((row) => {
|
||||
if (acceptableTypes.length > 0 && !acceptableTypes.includes(row.gpuType)) return false;
|
||||
if (row.gpuCount < requirement.gpuCount) return false;
|
||||
|
||||
if (requirement.requiresHighSpeedInterconnect) {
|
||||
const fast: InterconnectType[] = ['Infiniband', 'RoCE', 'NVLink'];
|
||||
if (!fast.includes(row.interconnectType)) return false;
|
||||
}
|
||||
if (requirement.minSecurityTier === 'secure_cloud' && row.securityTier !== 'secure_cloud') {
|
||||
return false;
|
||||
}
|
||||
if (requirement.startsAt && row.startsAt > requirement.startsAt) return false;
|
||||
if (requirement.endsAt && row.endsAt < requirement.endsAt) return false;
|
||||
if (requirement.totalGpuHours && row.availableGpuHours < requirement.totalGpuHours) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
})
|
||||
.map((row) => {
|
||||
const rationale: string[] = [];
|
||||
let score = 0.5;
|
||||
|
||||
// Prefer filling blocks that are sitting idle: the marginal hour on an
|
||||
// under-utilised commitment is already paid for, so selling it is worth
|
||||
// more than selling one from a block that is nearly full.
|
||||
const idleBonus = 1 - row.utilisation;
|
||||
score += idleBonus * 0.3;
|
||||
if (row.utilisation < 0.5) {
|
||||
rationale.push(
|
||||
`Block is ${Math.round(row.utilisation * 100)}% utilised — selling here reduces idle spend.`,
|
||||
);
|
||||
}
|
||||
|
||||
if (requirement.gpuType && row.gpuType === requirement.gpuType) {
|
||||
score += 0.1;
|
||||
rationale.push(`Exact GPU match (${row.gpuType}).`);
|
||||
} else if (acceptableTypes.includes(row.gpuType)) {
|
||||
rationale.push(`Acceptable alternative (${row.gpuType}).`);
|
||||
}
|
||||
|
||||
if (requirement.requiresHighSpeedInterconnect) {
|
||||
rationale.push(`${row.interconnectType} fabric meets the training requirement.`);
|
||||
}
|
||||
|
||||
// Margin headroom: can this be sold above break-even, within the
|
||||
// customer's ceiling?
|
||||
if (requirement.maxPricePerGpuHourCents && row.breakEvenPriceCents != null) {
|
||||
if (row.breakEvenPriceCents <= requirement.maxPricePerGpuHourCents) {
|
||||
score += 0.1;
|
||||
rationale.push(
|
||||
`Break-even is below the customer's ceiling — there is margin available.`,
|
||||
);
|
||||
} else {
|
||||
// Kept in the results but scored down, because a seller may still
|
||||
// want to know the option exists and price it as a loss leader.
|
||||
score -= 0.3;
|
||||
rationale.push(
|
||||
`⚠ Break-even exceeds the customer's stated ceiling. This would sell at a loss.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { ...row, score: Math.max(0, Math.min(1, score)), rationale };
|
||||
})
|
||||
.sort((a, b) => b.score - a.score);
|
||||
|
||||
return matches;
|
||||
}
|
||||
|
||||
/**
|
||||
* Blocks with meaningful unsold capacity — the alert that pays for PIG.
|
||||
*
|
||||
* Committed hours are already paid for, so idle capacity is money leaving the
|
||||
* business every hour it stays unsold. Filtered to blocks that are live or
|
||||
* starting soon, since idle capacity in eighteen months is a forecasting
|
||||
* matter rather than an alarm.
|
||||
*/
|
||||
async idleCapacity(options: { thresholdPct?: number; withinDays?: number } = {}) {
|
||||
const threshold = options.thresholdPct ?? 0.25;
|
||||
const horizon = new Date(Date.now() + (options.withinDays ?? 30) * 86_400_000);
|
||||
|
||||
const rows = await this.availability();
|
||||
return rows
|
||||
.filter((row) => row.startsAt <= horizon && 1 - row.utilisation >= threshold)
|
||||
.map((row) => {
|
||||
const idleHours = row.totalGpuHours - row.soldGpuHours;
|
||||
return {
|
||||
...row,
|
||||
idleGpuHours: idleHours,
|
||||
// The number that makes the case: what the unsold hours cost us.
|
||||
idleCostCents: Math.round(idleHours * row.costPerGpuHourCents),
|
||||
};
|
||||
})
|
||||
.sort((a, b) => b.idleCostCents - a.idleCostCents);
|
||||
}
|
||||
|
||||
/** Book-level margin across every live commitment. */
|
||||
async marginReport(): Promise<{
|
||||
totals: ReturnType<typeof computeMargin>;
|
||||
blocks: AvailabilityRow[];
|
||||
}> {
|
||||
const blocks = await this.availability();
|
||||
const allocRows = blocks.length
|
||||
? await this.db
|
||||
.select()
|
||||
.from(allocations)
|
||||
.where(
|
||||
and(
|
||||
inArray(
|
||||
allocations.capacityCommitmentId,
|
||||
blocks.map((b) => b.commitmentId),
|
||||
),
|
||||
inArray(allocations.status, [...CONSUMING_ALLOCATION_STATUSES]),
|
||||
),
|
||||
)
|
||||
: [];
|
||||
|
||||
const books = blocks.map((block) => ({
|
||||
commitment: {
|
||||
gpuHours: block.totalGpuHours,
|
||||
costPerGpuHourCents: block.costPerGpuHourCents,
|
||||
},
|
||||
allocations: allocRows
|
||||
.filter((a) => a.capacityCommitmentId === block.commitmentId)
|
||||
.map((a) => ({
|
||||
gpuHours: Number(a.gpuHours),
|
||||
pricePerGpuHourCents: a.pricePerGpuHourCents,
|
||||
})),
|
||||
}));
|
||||
|
||||
// Aggregate by summing cents, never by averaging per-block percentages —
|
||||
// an average of ratios weights a tiny block equally with a huge one.
|
||||
const { aggregateMargin } = await import('@pig/core');
|
||||
return { totals: aggregateMargin(books), blocks };
|
||||
}
|
||||
|
||||
/**
|
||||
* Release holds whose timer has lapsed.
|
||||
*
|
||||
* Availability already ignores expired holds, so this is bookkeeping rather
|
||||
* than correctness — it keeps the stored state honest and makes the release
|
||||
* visible in the UI.
|
||||
*/
|
||||
async sweepExpiredHolds(now = new Date()): Promise<number> {
|
||||
const result = await this.db
|
||||
.update(allocations)
|
||||
.set({ status: 'released', releasedAt: now, updatedAt: now })
|
||||
.where(
|
||||
and(
|
||||
eq(allocations.status, 'planned'),
|
||||
sql`${allocations.holdExpiresAt} IS NOT NULL`,
|
||||
lte(allocations.holdExpiresAt, now),
|
||||
),
|
||||
)
|
||||
.returning({ id: allocations.id });
|
||||
return result.length;
|
||||
}
|
||||
|
||||
/** Live inventory from providers, for capacity we do not yet hold. */
|
||||
async searchInventory(query: {
|
||||
gpuType?: string;
|
||||
minGpuCount?: number;
|
||||
maxPriceCents?: number;
|
||||
requiresHighSpeedInterconnect?: boolean;
|
||||
limit?: number;
|
||||
}) {
|
||||
const fast: InterconnectType[] = ['Infiniband', 'RoCE', 'NVLink'];
|
||||
return this.db
|
||||
.select()
|
||||
.from(inventoryListings)
|
||||
.where(
|
||||
and(
|
||||
query.gpuType ? eq(inventoryListings.gpuType, query.gpuType) : undefined,
|
||||
query.minGpuCount ? gte(inventoryListings.gpuCount, query.minGpuCount) : undefined,
|
||||
query.maxPriceCents
|
||||
? lte(inventoryListings.onDemandPriceCents, query.maxPriceCents)
|
||||
: undefined,
|
||||
query.requiresHighSpeedInterconnect
|
||||
? inArray(inventoryListings.interconnectType, fast)
|
||||
: undefined,
|
||||
// Unavailable stock is never a useful search result.
|
||||
or(
|
||||
eq(inventoryListings.stockStatus, 'Available'),
|
||||
eq(inventoryListings.stockStatus, 'Low'),
|
||||
eq(inventoryListings.stockStatus, 'Medium'),
|
||||
eq(inventoryListings.stockStatus, 'High'),
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(Math.min(query.limit ?? 50, 200));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
/**
|
||||
* Inventory sync from the Prime Intellect availability API.
|
||||
*
|
||||
* Runs on an interval, upserting listings on their natural key so repeated
|
||||
* syncs converge rather than accumulating near-duplicates.
|
||||
*
|
||||
* Two decisions worth explaining.
|
||||
*
|
||||
* **Stale listings are marked, not deleted.** A listing that disappears
|
||||
* upstream has usually sold out rather than ceased to exist, and deleting it
|
||||
* would destroy the price history that makes the inventory useful for
|
||||
* negotiation. Anything not seen in this pass is marked `Unavailable`, keeping
|
||||
* the record and its history while removing it from search.
|
||||
*
|
||||
* **A failed sync is not fatal.** The upstream publishes no rate limits, so
|
||||
* some failures are expected. The job logs and waits for the next tick rather
|
||||
* than crashing the server that people are using.
|
||||
*/
|
||||
import { and, eq, inArray, lt, sql } from 'drizzle-orm';
|
||||
import type { Database } from '@pig/db';
|
||||
import { inventoryListings } from '@pig/db';
|
||||
import { PrimeClient, mapListing } from '@pig/prime';
|
||||
import type { Config } from '../lib/config';
|
||||
|
||||
export function startPrimeSync(config: Config, db: Database): () => void {
|
||||
if (!config.PRIME_SYNC_ENABLED || !config.PRIME_API_KEY) {
|
||||
if (config.PRIME_SYNC_ENABLED) {
|
||||
console.warn('[pig] Prime sync enabled but no API key — not starting.');
|
||||
}
|
||||
return () => {};
|
||||
}
|
||||
|
||||
const client = new PrimeClient({
|
||||
apiKey: config.PRIME_API_KEY,
|
||||
baseUrl: config.PRIME_API_BASE,
|
||||
onRetry: ({ attempt, delayMs, reason }) =>
|
||||
console.warn(`[pig] prime retry ${attempt} in ${delayMs}ms (${reason})`),
|
||||
});
|
||||
|
||||
let running = false;
|
||||
|
||||
async function runOnce() {
|
||||
// Overlapping runs would fight over the same rows for no benefit.
|
||||
if (running) return;
|
||||
running = true;
|
||||
const startedAt = new Date();
|
||||
let upserted = 0;
|
||||
|
||||
try {
|
||||
for await (const page of client.iterateGpuAvailability()) {
|
||||
const rows = page
|
||||
.map(mapListing)
|
||||
.filter((row): row is NonNullable<typeof row> => row !== null);
|
||||
if (rows.length === 0) continue;
|
||||
|
||||
for (const row of rows) {
|
||||
await db
|
||||
.insert(inventoryListings)
|
||||
.values({
|
||||
...row,
|
||||
socket: row.socket as 'PCIe' | null,
|
||||
source: 'prime_api',
|
||||
observedAt: startedAt,
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
inventoryListings.externalCloudId,
|
||||
inventoryListings.gpuType,
|
||||
inventoryListings.socket,
|
||||
inventoryListings.gpuCount,
|
||||
inventoryListings.securityTier,
|
||||
],
|
||||
set: {
|
||||
stockStatus: row.stockStatus,
|
||||
onDemandPriceCents: row.onDemandPriceCents,
|
||||
communityPriceCents: row.communityPriceCents,
|
||||
priceIsVariable: row.priceIsVariable,
|
||||
interconnectGbps: row.interconnectGbps,
|
||||
interconnectType: row.interconnectType,
|
||||
provisioningMinutes: row.provisioningMinutes,
|
||||
region: row.region,
|
||||
country: row.country,
|
||||
images: row.images,
|
||||
raw: row.raw,
|
||||
observedAt: startedAt,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
});
|
||||
upserted += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// Anything the provider stopped listing is out of stock, not gone.
|
||||
const stale = await db
|
||||
.update(inventoryListings)
|
||||
.set({ stockStatus: 'Unavailable', updatedAt: new Date() })
|
||||
.where(
|
||||
and(
|
||||
eq(inventoryListings.source, 'prime_api'),
|
||||
lt(inventoryListings.observedAt, startedAt),
|
||||
sql`${inventoryListings.stockStatus} <> 'Unavailable'`,
|
||||
),
|
||||
)
|
||||
.returning({ id: inventoryListings.id });
|
||||
|
||||
console.log(
|
||||
`[pig] prime sync: ${upserted} listing(s) upserted, ${stale.length} marked unavailable`,
|
||||
);
|
||||
} catch (error) {
|
||||
// Logged, not thrown. A provider outage must not take down the CRM.
|
||||
console.error('[pig] prime sync failed:', error instanceof Error ? error.message : error);
|
||||
} finally {
|
||||
running = false;
|
||||
}
|
||||
}
|
||||
|
||||
// Delay the first run so startup is not competing with an outbound sync.
|
||||
const initial = setTimeout(() => void runOnce(), 10_000);
|
||||
const interval = setInterval(
|
||||
() => void runOnce(),
|
||||
config.PRIME_SYNC_INTERVAL_MINUTES * 60_000,
|
||||
);
|
||||
|
||||
console.log(`[pig] prime sync every ${config.PRIME_SYNC_INTERVAL_MINUTES}m`);
|
||||
|
||||
return () => {
|
||||
clearTimeout(initial);
|
||||
clearInterval(interval);
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user