Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / internal/platform/directory/directory.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 9bde0f9a74b89d01980fa658e5253acda7528d0c |
| Branch | per-tenant-routing |
| Path or archive member | internal/platform/directory/directory.go |
| Original SHA-256 | d96836a24f9b4e847166d5ac08b0d60436940e26093c851b6eb361894dbef04c |
| Git blob SHA | 84d37b23b5d9bbf1b33a36f38fa8690920bd73e8 |
| Reading-copy SHA-256 | d96836a24f9b4e847166d5ac08b0d60436940e26093c851b6eb361894dbef04c |
Every recorded occurrence
These 3 records contain identical original source bytes. Their paths and revisions remain part of the archive.
| Original record | Branch or archive | Revision / identifier | Original location |
|---|---|---|---|
| Janady13/missionmatch / internal/platform/directory/directory.go | per-tenant-routing | 9bde0f9a74b89d01980fa658e5253acda7528d0c | Open original |
| Janady13/missionmatch / internal/platform/directory/directory.go | deploy-and-ops | bdacd9d0b4d5998d25fc74bf4007f4c3ec538c7a | Open original |
| Janady13/missionmatch / internal/platform/directory/directory.go | per-tenant-routing | 9bde0f9a74b89d01980fa658e5253acda7528d0c | Open original |
Recovered source transcript
Download complete source text ↓168 lines · 6.8 KB
// Package directory reads and writes the tenant_directory catalog in the CONTROL PLANE database, the
// authoritative map from an organization to the isolated database that holds its business data under the
// database per tenant model. The request router calls Lookup after auth validates an organization to
// find its tenant database; the provisioning tools call Register, and the cross tenant migrator calls
// List. The directory holds only a credential REFERENCE (resolved through internal/platform/secrets at
// pool open time), never a plaintext credential.
//
// Read methods (Lookup, List) work for the NOBYPASSRLS app role (the routing path). Write methods
// (Register, SetStatus, SetMigrated) need the migrator role (provisioning and maintenance). The caller
// supplies the appropriately privileged control plane pool. The table is cross tenant and is NOT under
// the per request org GUC RLS, so a single connection sees every tenant's row (db names and references,
// never credentials), which is exactly what routing any authenticated tenant requires.
package directory
import (
"context"
"errors"
"fmt"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// ErrUnknownTenant means the organization has no row in the tenant directory, so it has no tenant
// database to route to. The router maps it to a fail closed denial, never a fallback to another tenant.
var ErrUnknownTenant = errors.New("directory: no tenant database for organization")
// Status is a tenant's lifecycle state in the directory (must match the CHECK in migration 0020).
type Status string
const (
StatusProvisioning Status = "provisioning"
StatusActive Status = "active"
StatusSuspended Status = "suspended"
StatusMigrating Status = "migrating"
StatusOffline Status = "offline"
)
// TenantConn is one directory row: where an organization's database lives and how to resolve its
// credential. SecretRef is a logical key for internal/platform/secrets, never a credential.
type TenantConn struct {
OrganizationID uuid.UUID
DBName string
Host string
Port int
SecretRef string
Status Status
Cloud string
Region string
LastMigratedVersion *int64
LastMigratedAt *time.Time
}
// Directory accesses the tenant_directory catalog over a control plane pool.
type Directory struct{ pool *pgxpool.Pool }
// New builds a Directory over a control plane pool. Pass the app pool for read only routing, or the
// migrator pool for provisioning and maintenance writes.
func New(pool *pgxpool.Pool) *Directory { return &Directory{pool: pool} }
const selectCols = `organization_id, db_name, db_host, db_port, connection_secret_ref, status, cloud, region, last_migrated_version, last_migrated_at`
func scan(row pgx.Row) (TenantConn, error) {
var t TenantConn
if err := row.Scan(&t.OrganizationID, &t.DBName, &t.Host, &t.Port, &t.SecretRef, &t.Status, &t.Cloud, &t.Region, &t.LastMigratedVersion, &t.LastMigratedAt); err != nil {
return TenantConn{}, err
}
return t, nil
}
// Lookup returns the directory row for an organization, or ErrUnknownTenant if there is none. It does
// NOT filter by status: the router decides whether a non active status may serve a request.
func (d *Directory) Lookup(ctx context.Context, orgID uuid.UUID) (TenantConn, error) {
t, err := scan(d.pool.QueryRow(ctx, `SELECT `+selectCols+` FROM tenant_directory WHERE organization_id = $1`, orgID))
if errors.Is(err, pgx.ErrNoRows) {
return TenantConn{}, fmt.Errorf("%w: %s", ErrUnknownTenant, orgID)
}
if err != nil {
return TenantConn{}, fmt.Errorf("directory lookup: %w", err)
}
return t, nil
}
// List returns directory rows in the given statuses (no statuses means all), ordered by db_name for a
// stable fan out (the cross tenant migrator iterates this).
func (d *Directory) List(ctx context.Context, statuses ...Status) ([]TenantConn, error) {
q := `SELECT ` + selectCols + ` FROM tenant_directory`
var args []any
if len(statuses) > 0 {
ss := make([]string, len(statuses))
for i, s := range statuses {
ss[i] = string(s)
}
q += ` WHERE status = ANY($1)`
args = append(args, ss)
}
q += ` ORDER BY db_name`
rows, err := d.pool.Query(ctx, q, args...)
if err != nil {
return nil, fmt.Errorf("directory list: %w", err)
}
defer rows.Close()
var out []TenantConn
for rows.Next() {
t, err := scan(rows)
if err != nil {
return nil, err
}
out = append(out, t)
}
return out, rows.Err()
}
// Register inserts or updates a tenant's directory row (migrator privilege). It is how provisioning
// publishes a new tenant database, and is idempotent: a re register updates the connection info and
// status without disturbing the migration bookkeeping.
func (d *Directory) Register(ctx context.Context, t TenantConn) error {
if t.Status == "" {
t.Status = StatusProvisioning
}
if t.Port == 0 {
t.Port = 5432
}
if t.Cloud == "" {
t.Cloud = "local"
}
_, err := d.pool.Exec(ctx,
`INSERT INTO tenant_directory (organization_id, db_name, db_host, db_port, connection_secret_ref, status, cloud, region)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
ON CONFLICT (organization_id) DO UPDATE SET
db_name = EXCLUDED.db_name, db_host = EXCLUDED.db_host, db_port = EXCLUDED.db_port,
connection_secret_ref = EXCLUDED.connection_secret_ref, status = EXCLUDED.status,
cloud = EXCLUDED.cloud, region = EXCLUDED.region, updated_at = now()`,
t.OrganizationID, t.DBName, t.Host, t.Port, t.SecretRef, t.Status, t.Cloud, t.Region)
if err != nil {
return fmt.Errorf("directory register: %w", err)
}
return nil
}
// SetStatus flips a tenant's lifecycle status (migrator privilege), for example to migrating or offline
// during a move or maintenance and back to active when it completes.
func (d *Directory) SetStatus(ctx context.Context, orgID uuid.UUID, status Status) error {
tag, err := d.pool.Exec(ctx, `UPDATE tenant_directory SET status = $2, updated_at = now() WHERE organization_id = $1`, orgID, status)
if err != nil {
return fmt.Errorf("directory set status: %w", err)
}
if tag.RowsAffected() == 0 {
return fmt.Errorf("%w: %s", ErrUnknownTenant, orgID)
}
return nil
}
// SetMigrated records the last applied schema version for a tenant after a successful migrate (migrator
// privilege), so the cross tenant migrator can report which databases are at head.
func (d *Directory) SetMigrated(ctx context.Context, orgID uuid.UUID, version int64) error {
tag, err := d.pool.Exec(ctx, `UPDATE tenant_directory SET last_migrated_version = $2, last_migrated_at = now(), updated_at = now() WHERE organization_id = $1`, orgID, version)
if err != nil {
return fmt.Errorf("directory set migrated: %w", err)
}
if tag.RowsAffected() == 0 {
return fmt.Errorf("%w: %s", ErrUnknownTenant, orgID)
}
return nil
}