feat(data): add PostgresSource for live config reads

go.mod

@@ -17,6 +17,8 @@ require (
 	github.com/goccy/go-json v0.10.5
 	github.com/google/uuid v1.6.0
 	github.com/invopop/jsonschema v0.13.0
+	github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438
+	github.com/jackc/pgx/v5 v5.10.0
 	github.com/kelindar/bitmap v1.5.3
 	github.com/nats-io/nats-server/v2 v2.12.6
 	github.com/nats-io/nats.go v1.49.0
@@ -64,6 +66,9 @@ require (
 	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/kelindar/simd v1.1.2 // indirect
 	github.com/klauspost/compress v1.18.4 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
@@ -84,6 +89,7 @@ require (
 	golang.org/x/crypto v0.49.0 // indirect
 	golang.org/x/exp v0.0.0-20250911091902-df9299821621 // indirect
 	golang.org/x/net v0.51.0 // indirect
+	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.35.0 // indirect
 	golang.org/x/time v0.15.0 // indirect

go.sum

@@ -94,6 +94,16 @@ github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/invopop/jsonschema v0.13.0 h1:KvpoAJWEjR3uD9Kbm2HWJmqsEaHt8lBUpd0qHcIi21E=
 github.com/invopop/jsonschema v0.13.0/go.mod h1:ffZ5Km5SWWRAIN6wbDXItl95euhFz2uON45H2qjYt+0=
+github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0=
+github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/kelindar/bitmap v1.5.3 h1:/ty1SvbLE5ZKO4ToFNeXe3P3RrQsoj4a0x5gZNp5Vzo=
 github.com/kelindar/bitmap v1.5.3/go.mod h1:j3qZjxH9s4OtvsnFTP2bmPkjqil9Y2xQlxPYHexasEA=
@@ -148,6 +158,8 @@ github.com/stoewer/go-strcase v1.3.1/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8w
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
@@ -181,6 +193,8 @@ golang.org/x/exp v0.0.0-20250911091902-df9299821621 h1:2id6c1/gto0kaHYyrixvknJ8t
 golang.org/x/exp v0.0.0-20250911091902-df9299821621/go.mod h1:TwQYMMnGpvZyc+JpB/UAuTNIsVJifOlSkrZkhcvpVUk=
 golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
 golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

pkg/plugin/data/plugin.go

@@ -56,6 +56,10 @@ type (
 	// Source after the primary JSON has been unmarshaled.
 	Resolver = system.Resolver
 
+	// Singleton is an optional Definition marker declaring the kind's config is a single object
+	// rather than a collection — its source returns one JSON object instead of an array.
+	Singleton = system.Singleton
+
 	// Validator is an optional Definition interface for enforcing post-load invariants. A
 	// non-nil error panics.
 	Validator = system.Validator
@@ -121,6 +125,19 @@ func (p *Plugin) Source() Source {
 func Register[T Definition](p *Plugin) {
 	var zero T
 	p.state.AddKind(zero.Name(), zero.JSONFile(), system.MakeAssemble[T]())
+
+	// Tell the source how to read this kind, when the source supports it. EmbedSource and the test
+	// fakes don't (they key off the file path), so these are no-ops for them; a Postgres-backed
+	// source records the file→table mapping and, for a Singleton kind, that the table is read as a
+	// single object instead of an array.
+	if r, ok := p.config.Source.(system.KindRegistrar); ok {
+		r.RegisterKind(zero.JSONFile(), zero.Name())
+	}
+	if _, isSingleton := any(zero).(system.Singleton); isSingleton {
+		if r, ok := p.config.Source.(system.SingletonRegistrar); ok {
+			r.RegisterSingleton(zero.JSONFile())
+		}
+	}
 }
 
 // registered is the process-global plugin instance set by Plugin.Register(world). It lets

pkg/plugin/data/system/definition.go

@@ -26,6 +26,17 @@ type Definition interface {
 	JSONFile() string
 }
 
+// Singleton is an optional marker a Definition may implement to declare that its config is a single
+// object rather than a collection of records. A singleton kind's source returns one JSON object
+// (e.g. {"maxPlayers":8,"roundSeconds":90}) and its backing table holds at most one row, whereas a
+// non-singleton kind's source returns a JSON array of records.
+//
+// SingleObject is a pure marker — it is never called. Implementing it (on a value receiver, like
+// Name/JSONFile) is the entire signal.
+type Singleton interface {
+	SingleObject()
+}
+
 // Resolver is an optional interface a Definition may implement to perform a second-stage load
 // after the primary JSON has been unmarshaled — typically to fetch additional designer-bundled
 // files the JSON references (e.g. Tiled .tmj tilemaps that map_levels.json points at).

pkg/plugin/data/system/postgres.go

@@ -0,0 +1,137 @@
+package system
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"path"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/rotisserie/eris"
+)
+
+// PostgresSource reads each config kind's rows from a Postgres database as JSON bytes. Every config
+// table has the uniform shape (id text PRIMARY KEY, doc jsonb): one row per record, the record's
+// JSON in doc. Like EmbedSource it is single-version: it ignores the requested hash and returns the
+// current rows, so a snapshot restore resumes on current config (Reconcile's warn path) instead of
+// erroring (the panic path).
+//
+// A kind may be registered as a singleton (RegisterSingleton) — its table holds at most one row and
+// Fetch returns one JSON object instead of an array — and may be given an explicit table name
+// (RegisterKind). Unregistered kinds fall back to a table name derived from the JSON file.
+type PostgresSource struct {
+	reader     configReader
+	singletons map[string]bool   // jsonFile → read as a single object instead of an array
+	tables     map[string]string // jsonFile → table name (overrides tableFromFile)
+}
+
+var (
+	_ Source             = (*PostgresSource)(nil)
+	_ SingletonRegistrar = (*PostgresSource)(nil)
+	_ KindRegistrar      = (*PostgresSource)(nil)
+)
+
+// configReader is the DB seam, extracted so Fetch is unit-testable without a live database.
+type configReader interface {
+	// readTableJSON returns table's rows as JSON: a single object when singleton is true (the first
+	// row), otherwise a deterministic JSON array ("[]" when empty).
+	readTableJSON(ctx context.Context, table string, singleton bool) ([]byte, error)
+}
+
+// SingletonRegistrar is implemented by sources that can be told a kind's config is a single object
+// rather than a collection of records.
+type SingletonRegistrar interface {
+	RegisterSingleton(file string)
+}
+
+// KindRegistrar is implemented by sources that accept an explicit table name for a kind's JSON file.
+type KindRegistrar interface {
+	RegisterKind(file, table string)
+}
+
+// NewPostgresSource opens a pgx pool against dsn (a read-only config-database DSN). The pool
+// connects lazily: a malformed dsn fails here, connectivity fails on the first Fetch.
+func NewPostgresSource(ctx context.Context, dsn string) (*PostgresSource, error) {
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		return nil, eris.Wrap(err, "data: opening config postgres pool")
+	}
+	return &PostgresSource{
+		reader:     pgxReader{pool: pool},
+		singletons: map[string]bool{},
+		tables:     map[string]string{},
+	}, nil
+}
+
+// RegisterSingleton marks file's kind as a single-object config: Fetch returns one JSON object and
+// the backing table is expected to hold at most one row.
+func (p *PostgresSource) RegisterSingleton(file string) {
+	if p.singletons == nil {
+		p.singletons = map[string]bool{}
+	}
+	p.singletons[file] = true
+}
+
+// RegisterKind records the table name to read file's kind from, overriding the default derived from
+// the file name.
+func (p *PostgresSource) RegisterKind(file, table string) {
+	if p.tables == nil {
+		p.tables = map[string]string{}
+	}
+	p.tables[file] = table
+}
+
+// tableFor returns the table registered for file, falling back to the name derived from the file.
+func (p *PostgresSource) tableFor(file string) string {
+	if t, ok := p.tables[file]; ok {
+		return t
+	}
+	return tableFromFile(file)
+}
+
+// Fetch returns the current contents of file's table as JSON plus their sha256 hex. hash is ignored.
+// A missing or empty table surfaces as a read error and propagates (fail loud) — config is read from
+// Postgres only.
+func (p *PostgresSource) Fetch(ctx context.Context, file, _ string) ([]byte, string, error) {
+	table := p.tableFor(file)
+	raw, err := p.reader.readTableJSON(ctx, table, p.singletons[file])
+	if err != nil {
+		return nil, "", eris.Wrapf(err, "data: postgres source reading table %q for %q", table, file)
+	}
+	sum := sha256.Sum256(raw)
+	return raw, hex.EncodeToString(sum[:]), nil
+}
+
+// tableFromFile maps a kind's JSONFile() ("testdata/abilities.json") to its table ("abilities").
+func tableFromFile(file string) string {
+	base := path.Base(file)
+	return strings.TrimSuffix(base, path.Ext(base))
+}
+
+type pgxReader struct {
+	pool *pgxpool.Pool
+}
+
+// readTableJSON reads table's doc column as JSON. For a singleton it returns the first row's doc as a
+// JSON object; otherwise it aggregates every row's doc into one array ordered by id for a stable
+// hash. table is a trusted kind identifier but is quoted via pgx.Identifier.
+//
+// A missing table or an empty singleton table surfaces as a query error (SQLSTATE 42P01 /
+// pgx.ErrNoRows respectively) and is left to propagate — there is no fallback, so an unreadable
+// table is a genuine boot failure.
+func (r pgxReader) readTableJSON(ctx context.Context, table string, singleton bool) ([]byte, error) {
+	ident := pgx.Identifier{table}.Sanitize()
+	var query string
+	if singleton {
+		query = "SELECT doc::text FROM " + ident + " AS t LIMIT 1"
+	} else {
+		query = "SELECT coalesce(json_agg(doc ORDER BY id), '[]'::json)::text FROM " + ident + " AS t"
+	}
+	var out string
+	if err := r.pool.QueryRow(ctx, query).Scan(&out); err != nil {
+		return nil, err
+	}
+	return []byte(out), nil
+}

pkg/plugin/data/system/postgres_internal_test.go

@@ -0,0 +1,147 @@
+package system
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// fakeReader is a configReader stub for exercising the Fetch contract without a database. rows holds
+// array-mode payloads and singleton holds single-object payloads, both keyed by table; the reader
+// picks between them on the singleton flag and records every call.
+type fakeReader struct {
+	rows       map[string][]byte
+	singleton  map[string][]byte
+	err        error
+	calls      []string
+	lastSingle bool
+}
+
+func (f *fakeReader) readTableJSON(_ context.Context, table string, singleton bool) ([]byte, error) {
+	f.calls = append(f.calls, table)
+	f.lastSingle = singleton
+	if f.err != nil {
+		return nil, f.err
+	}
+	if singleton {
+		return f.singleton[table], nil
+	}
+	return f.rows[table], nil
+}
+
+func sha256Hex(b []byte) string {
+	sum := sha256.Sum256(b)
+	return hex.EncodeToString(sum[:])
+}
+
+func TestTableFromFile(t *testing.T) {
+	cases := map[string]string{
+		"testdata/abilities.json": "abilities",
+		"abilities.json":          "abilities",
+		"data/mobs.json":          "mobs",
+		"loot_table.json":         "loot_table",
+	}
+	for in, want := range cases {
+		require.Equalf(t, want, tableFromFile(in), "tableFromFile(%q)", in)
+	}
+}
+
+func TestPostgresSourceFetchReturnsRowsAndHash(t *testing.T) {
+	body := []byte(`[{"id":"fireball","cooldown":3}]`)
+	src := &PostgresSource{
+		reader: &fakeReader{rows: map[string][]byte{"abilities": body}},
+	}
+
+	got, hash, err := src.Fetch(context.Background(), "testdata/abilities.json", "")
+
+	require.NoError(t, err)
+	require.Equal(t, body, got)
+	require.Equal(t, sha256Hex(body), hash, "hash must be sha256 of the returned bytes, matching EmbedSource")
+}
+
+// Single-version: an unservable hash returns current rows (no error), so restore takes the warn
+// path, not the panic path.
+func TestPostgresSourceFetchIgnoresRequestedHash(t *testing.T) {
+	body := []byte(`[{"id":"a"}]`)
+	src := &PostgresSource{
+		reader: &fakeReader{rows: map[string][]byte{"abilities": body}},
+	}
+
+	got, hash, err := src.Fetch(context.Background(), "abilities.json", "some-old-snapshot-hash")
+
+	require.NoError(t, err, "an unservable historical hash must NOT error (single-version semantics)")
+	require.Equal(t, body, got)
+	require.Equal(t, sha256Hex(body), hash)
+	require.NotEqual(t, "some-old-snapshot-hash", hash)
+}
+
+func TestPostgresSourceFetchResolvesTableFromFile(t *testing.T) {
+	fr := &fakeReader{rows: map[string][]byte{"mobs": []byte("[]")}}
+	src := &PostgresSource{reader: fr}
+
+	_, _, err := src.Fetch(context.Background(), "config/mobs.json", "")
+
+	require.NoError(t, err)
+	require.Equal(t, []string{"mobs"}, fr.calls, "Fetch must query the table derived from the file name")
+}
+
+func TestPostgresSourceFetchWrapsReaderError(t *testing.T) {
+	sentinel := errors.New("connection refused")
+	src := &PostgresSource{
+		reader: &fakeReader{err: sentinel},
+	}
+
+	_, _, err := src.Fetch(context.Background(), "abilities.json", "")
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, sentinel, "reader errors must propagate (a boot-time load failure), not be swallowed")
+}
+
+// A singleton kind reads as a single JSON object via the singleton query path, not the array
+// aggregate, and hashes the returned object bytes like every other source.
+func TestPostgresSourceFetchSingletonReturnsObject(t *testing.T) {
+	obj := []byte(`{"maxPlayers":8,"roundSeconds":90}`)
+	fr := &fakeReader{singleton: map[string][]byte{"match_settings": obj}}
+	src := &PostgresSource{reader: fr}
+	src.RegisterSingleton("match_settings.json")
+
+	got, hash, err := src.Fetch(context.Background(), "match_settings.json", "")
+
+	require.NoError(t, err)
+	require.True(t, fr.lastSingle, "a registered singleton must take the single-object read path")
+	require.Equal(t, obj, got, "singleton Fetch must return one JSON object, not an array")
+	require.Equal(t, sha256Hex(obj), hash)
+}
+
+// A kind that was not registered as a singleton must take the array read path.
+func TestPostgresSourceFetchNonSingletonReadsArray(t *testing.T) {
+	arr := []byte(`[{"id":"a"}]`)
+	fr := &fakeReader{rows: map[string][]byte{"abilities": arr}}
+	src := &PostgresSource{reader: fr}
+
+	got, _, err := src.Fetch(context.Background(), "abilities.json", "")
+
+	require.NoError(t, err)
+	require.False(t, fr.lastSingle, "an unregistered kind must take the array read path")
+	require.Equal(t, arr, got)
+}
+
+// RegisterKind maps file→table so tableFor returns the registered table; unregistered files fall
+// back to the name derived from the file.
+func TestPostgresSourceTableFor(t *testing.T) {
+	src := &PostgresSource{}
+
+	require.Equal(t, "abilities", src.tableFor("testdata/abilities.json"),
+		"an unregistered file must fall back to tableFromFile")
+
+	src.RegisterKind("powerups.json", "powerups_definitions")
+
+	require.Equal(t, "powerups_definitions", src.tableFor("powerups.json"),
+		"a registered file must resolve to its explicit table")
+	require.Equal(t, "mobs", src.tableFor("data/mobs.json"),
+		"other files keep falling back to tableFromFile")
+}