package main import ( "context" "encoding/json" "errors" "fmt" "sort" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgconn" ) type pgDatabase struct { pool *pgxpool.Pool } type pgCollection struct { db *pgDatabase spec relationalResourceSpec } type pgCursor struct { docs []Doc index int } type pgSingleResult struct { doc Doc err error } type pgDeleteResult struct { DeletedCount int64 } type pgUpdateResult struct { ModifiedCount int64 } type pgInsertResult struct{} type pgReplaceResult struct{} func openPostgres(ctx context.Context, databaseURL string) (*pgDatabase, error) { if strings.TrimSpace(databaseURL) == "" { return nil, errors.New("DATABASE_URL is required") } pool, err := pgxpool.New(ctx, databaseURL) if err != nil { return nil, fmt.Errorf("configure postgres: %w", err) } db := &pgDatabase{pool: pool} if err := db.Ping(ctx); err != nil { pool.Close() return nil, fmt.Errorf("ping postgres: %w", err) } if err := db.ensureRelationalSchema(ctx); err != nil { pool.Close() return nil, fmt.Errorf("initialize relational postgres schema: %w", err) } return db, nil } func (d *pgDatabase) Ping(ctx context.Context) error { return d.pool.Ping(ctx) } func (d *pgDatabase) Close() { if d != nil && d.pool != nil { d.pool.Close() } } func (d *pgDatabase) Collection(name string) *pgCollection { spec, ok := relationalResources[name] if !ok { return &pgCollection{db: d, spec: relationalResourceSpec{Name: name, Table: name}} } return &pgCollection{db: d, spec: spec} } func (c *pgCollection) Name() string { return c.spec.Name } func (c *pgCollection) valid() error { if _, ok := relationalResources[c.spec.Name]; !ok { return fmt.Errorf("unknown relational resource %s", c.spec.Name) } return nil } func (c *pgCollection) Find(ctx context.Context, filter Doc, opts ...*FindOptions) (*pgCursor, error) { if err := c.valid(); err != nil { return nil, err } args := []any{} where, err := c.buildFilter(filter, &args) if err != nil { return nil, err } query := "SELECT " + c.documentExpression() + " FROM " + relationalIdent(c.spec.Table) + " WHERE " + where if len(opts) > 0 && opts[0] != nil { query += c.sortClause(opts[0].Sort) if opts[0].Limit != nil && *opts[0].Limit > 0 { args = append(args, *opts[0].Limit) query += fmt.Sprintf(" LIMIT $%d", len(args)) } } rows, err := c.db.pool.Query(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() docs := make([]Doc, 0) for rows.Next() { var raw []byte if err := rows.Scan(&raw); err != nil { return nil, err } doc, err := decodeRelationalDocument(raw) if err != nil { return nil, err } docs = append(docs, doc) } if err := rows.Err(); err != nil { return nil, err } return &pgCursor{docs: docs}, nil } func (c *pgCollection) FindOne(ctx context.Context, filter Doc, opts ...*FindOneOptions) *pgSingleResult { if err := c.valid(); err != nil { return &pgSingleResult{err: err} } args := []any{} where, err := c.buildFilter(filter, &args) if err != nil { return &pgSingleResult{err: err} } query := "SELECT " + c.documentExpression() + " FROM " + relationalIdent(c.spec.Table) + " WHERE " + where if len(opts) > 0 && opts[0] != nil { query += c.sortClause(opts[0].Sort) } query += " LIMIT 1" var raw []byte if err := c.db.pool.QueryRow(ctx, query, args...).Scan(&raw); err != nil { if errors.Is(err, pgx.ErrNoRows) { err = ErrNotFound } return &pgSingleResult{err: err} } doc, err := decodeRelationalDocument(raw) return &pgSingleResult{doc: doc, err: err} } func (c *pgCollection) CountDocuments(ctx context.Context, filter Doc, _ ...*CountOptions) (int64, error) { if err := c.valid(); err != nil { return 0, err } args := []any{} where, err := c.buildFilter(filter, &args) if err != nil { return 0, err } var count int64 if err := c.db.pool.QueryRow(ctx, "SELECT COUNT(*) FROM "+c.spec.Table+" WHERE "+where, args...).Scan(&count); err != nil { return 0, err } return count, nil } func (c *pgCollection) InsertOne(ctx context.Context, doc any, _ ...*InsertOneOptions) (*pgInsertResult, error) { if err := c.valid(); err != nil { return nil, err } normalized, err := normalizeRelationalDocument(doc) if err != nil { return nil, err } tx, err := c.db.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) if err := c.writeDocument(ctx, tx, normalized, false); err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return &pgInsertResult{}, nil } func (c *pgCollection) ReplaceOne(ctx context.Context, filter any, replacement any, _ ...*ReplaceOptions) (*pgReplaceResult, error) { if err := c.valid(); err != nil { return nil, err } normalized, err := normalizeRelationalDocument(replacement) if err != nil { return nil, err } if _, ok := normalized["id"]; !ok { if filterMap, ok := asDocMap(filter); ok { if id, exists := filterMap["id"]; exists { normalized["id"] = id } } } tx, err := c.db.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) if err := c.writeDocument(ctx, tx, normalized, true); err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return &pgReplaceResult{}, nil } func (c *pgCollection) DeleteOne(ctx context.Context, filter Doc, _ ...*DeleteOptions) (*pgDeleteResult, error) { if err := c.valid(); err != nil { return nil, err } args := []any{} where, err := c.buildFilter(filter, &args) if err != nil { return nil, err } query := "DELETE FROM " + relationalIdent(c.spec.Table) + " WHERE " + relationalIdent("id") + " = (SELECT " + relationalIdent("id") + " FROM " + relationalIdent(c.spec.Table) + " WHERE " + where + " LIMIT 1)" tag, err := c.db.pool.Exec(ctx, query, args...) if err != nil { return nil, err } return &pgDeleteResult{DeletedCount: tag.RowsAffected()}, nil } func (c *pgCollection) DeleteMany(ctx context.Context, filter Doc, _ ...*DeleteOptions) (*pgDeleteResult, error) { if err := c.valid(); err != nil { return nil, err } args := []any{} where, err := c.buildFilter(filter, &args) if err != nil { return nil, err } tag, err := c.db.pool.Exec(ctx, "DELETE FROM "+relationalIdent(c.spec.Table)+" WHERE "+where, args...) if err != nil { return nil, err } return &pgDeleteResult{DeletedCount: tag.RowsAffected()}, nil } func (c *pgCollection) FindOneAndUpdate(ctx context.Context, filter any, update any, _ ...*FindOneAndUpdateOptions) *pgSingleResult { if err := c.valid(); err != nil { return &pgSingleResult{err: err} } filterMap, ok := asDocMap(filter) if !ok { return &pgSingleResult{err: errors.New("invalid filter")} } updateMap, ok := asDocMap(update) if !ok { return &pgSingleResult{err: errors.New("invalid update")} } tx, err := c.db.pool.Begin(ctx) if err != nil { return &pgSingleResult{err: err} } defer tx.Rollback(ctx) args := []any{} where, err := c.buildFilter(filterMap, &args) if err != nil { return &pgSingleResult{err: err} } var raw []byte query := "SELECT " + c.documentExpression() + " FROM " + relationalIdent(c.spec.Table) + " WHERE " + where + " LIMIT 1 FOR UPDATE" if err := tx.QueryRow(ctx, query, args...).Scan(&raw); err != nil { if errors.Is(err, pgx.ErrNoRows) { err = ErrNotFound } return &pgSingleResult{err: err} } doc, err := decodeRelationalDocument(raw) if err != nil { return &pgSingleResult{err: err} } applyRelationalUpdate(doc, updateMap) if err := c.writeDocument(ctx, tx, doc, true); err != nil { return &pgSingleResult{err: err} } if err := tx.Commit(ctx); err != nil { return &pgSingleResult{err: err} } return &pgSingleResult{doc: doc} } func (c *pgCollection) UpdateMany(ctx context.Context, filter any, update any, _ ...*UpdateOptions) (*pgUpdateResult, error) { if err := c.valid(); err != nil { return nil, err } filterMap, ok := asDocMap(filter) if !ok { return nil, errors.New("invalid filter") } updateMap, ok := asDocMap(update) if !ok { return nil, errors.New("invalid update") } tx, err := c.db.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) args := []any{} where, err := c.buildFilter(filterMap, &args) if err != nil { return nil, err } rows, err := tx.Query(ctx, "SELECT "+c.documentExpression()+" FROM "+relationalIdent(c.spec.Table)+" WHERE "+where+" FOR UPDATE", args...) if err != nil { return nil, err } docs := make([]Doc, 0) for rows.Next() { var raw []byte if err := rows.Scan(&raw); err != nil { rows.Close() return nil, err } doc, err := decodeRelationalDocument(raw) if err != nil { rows.Close() return nil, err } docs = append(docs, doc) } if err := rows.Err(); err != nil { rows.Close() return nil, err } rows.Close() var modified int64 for _, doc := range docs { applyRelationalUpdate(doc, updateMap) if err := c.writeDocument(ctx, tx, doc, true); err != nil { return nil, err } modified++ } if err := tx.Commit(ctx); err != nil { return nil, err } return &pgUpdateResult{ModifiedCount: modified}, nil } type relationalExecer interface { Exec(context.Context, string, ...any) (interface{ RowsAffected() int64 }, error) } type pgxExecer interface { Exec(context.Context, string, ...any) (pgconnCommandTag, error) } type pgconnCommandTag interface { RowsAffected() int64 } func (c *pgCollection) writeDocument(ctx context.Context, exec interface { Exec(context.Context, string, ...any) (pgconn.CommandTag, error) }, doc Doc, upsert bool) error { if err := c.validateDocument(doc); err != nil { return err } id := strings.TrimSpace(documentString(doc["id"])) if id == "" { return errors.New("document id is required") } fields := c.spec.columnFields() columns := make([]string, 0, len(fields)) placeholders := make([]string, 0, len(fields)) args := make([]any, 0, len(fields)) updateParts := make([]string, 0, len(fields)) for _, field := range fields { value, exists := doc[field] if !exists { value = nil if field == "createdAt" || field == "updatedAt" { value = int64(0) } } column := relationalColumnName(field) columns = append(columns, relationalIdent(column)) args = append(args, relationalDBValue(c.spec, field, value)) placeholder := fmt.Sprintf("$%d", len(args)) if c.spec.isJSON(field) { placeholder += "::jsonb" } placeholders = append(placeholders, placeholder) if field != "id" { updateParts = append(updateParts, relationalIdent(column)+" = EXCLUDED."+relationalIdent(column)) } } if len(columns) == 0 { return errors.New("id must be present in relational write") } query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", relationalIdent(c.spec.Table), strings.Join(columns, ", "), strings.Join(placeholders, ", ")) if upsert { query += " ON CONFLICT (" + relationalIdent("id") + ") DO UPDATE SET " + strings.Join(updateParts, ", ") } if _, err := exec.Exec(ctx, query, args...); err != nil { return err } return c.syncListRelations(ctx, exec, doc) } func (c *pgCollection) validateDocument(doc Doc) error { for field, value := range doc { if strings.HasPrefix(field, "_") { continue } if c.spec.hasField(field) { continue } if value == nil { continue } return fmt.Errorf("field %q is not mapped for relational resource %s", field, c.spec.Name) } return nil } func (c *pgCollection) buildFilter(filter Doc, args *[]any) (string, error) { if len(filter) == 0 { return "TRUE", nil } keys := make([]string, 0, len(filter)) for key := range filter { keys = append(keys, key) } sort.Strings(keys) clauses := make([]string, 0, len(filter)) for _, field := range keys { raw := filter[field] if field == "$or" { items, ok := asAnySlice(raw) if !ok || len(items) == 0 { return "", errors.New("$or requires a non-empty array") } parts := make([]string, 0, len(items)) for _, item := range items { sub, ok := asDocMap(item) if !ok { return "", errors.New("$or entries must be objects") } clause, err := c.buildFilter(sub, args) if err != nil { return "", err } parts = append(parts, "("+clause+")") } clauses = append(clauses, "("+strings.Join(parts, " OR ")+")") continue } if relation, ok := c.spec.ListRelations[field]; ok { clause, err := c.listRelationFilterClause(relation, raw, args) if err != nil { return "", err } clauses = append(clauses, clause) continue } if !c.spec.hasColumn(field) { return "", fmt.Errorf("field %q is not queryable on %s", field, c.spec.Name) } column := relationalIdent(relationalColumnName(field)) if operators, ok := asDocMap(raw); ok && containsRelationalOperator(operators) { for operator, operand := range operators { clause, err := c.operatorClause(column, field, operator, operand, args) if err != nil { return "", err } clauses = append(clauses, clause) } continue } argsValue := relationalDBValue(c.spec, field, raw) *args = append(*args, argsValue) if c.spec.isJSON(field) { clauses = append(clauses, fmt.Sprintf("%s = $%d::jsonb", column, len(*args))) } else { clauses = append(clauses, fmt.Sprintf("%s = $%d", column, len(*args))) } } if len(clauses) == 0 { return "TRUE", nil } return strings.Join(clauses, " AND "), nil } func (c *pgCollection) operatorClause(column, field, operator string, operand any, args *[]any) (string, error) { switch operator { case "$exists": exists, ok := operand.(bool) if !ok { return "", errors.New("$exists requires a boolean") } if exists { return column + " IS NOT NULL", nil } return column + " IS NULL", nil case "$ne": *args = append(*args, relationalDBValue(c.spec, field, operand)) if c.spec.isJSON(field) { return fmt.Sprintf("COALESCE(%s <> $%d::jsonb, TRUE)", column, len(*args)), nil } return fmt.Sprintf("COALESCE(%s <> $%d, TRUE)", column, len(*args)), nil case "$in": values, ok := asAnySlice(operand) if !ok || len(values) == 0 { return "FALSE", nil } parts := make([]string, 0, len(values)) for _, value := range values { *args = append(*args, relationalDBValue(c.spec, field, value)) if c.spec.isJSON(field) { parts = append(parts, fmt.Sprintf("(%s = $%d::jsonb OR %s @> jsonb_build_array($%d::jsonb))", column, len(*args), column, len(*args))) } else { parts = append(parts, fmt.Sprintf("%s = $%d", column, len(*args))) } } return "(" + strings.Join(parts, " OR ") + ")", nil case "$gt", "$gte", "$lt", "$lte": if c.spec.isJSON(field) { return "", fmt.Errorf("ordered comparison is not supported for JSON field %s", field) } op := map[string]string{"$gt": ">", "$gte": ">=", "$lt": "<", "$lte": "<="}[operator] *args = append(*args, relationalDBValue(c.spec, field, operand)) return fmt.Sprintf("%s %s $%d", column, op, len(*args)), nil default: return "", fmt.Errorf("unsupported relational operator %s", operator) } } func (c *pgCollection) sortClause(raw any) string { sortDoc, ok := raw.(Sort) if !ok || len(sortDoc) == 0 { return "" } parts := make([]string, 0, len(sortDoc)) for _, item := range sortDoc { if !c.spec.hasColumn(item.Key) || c.spec.isJSON(item.Key) { continue } direction := "ASC" if number, ok := numericValue(item.Value); ok && number < 0 { direction = "DESC" } parts = append(parts, relationalIdent(relationalColumnName(item.Key))+" "+direction+" NULLS LAST") } if len(parts) == 0 { return "" } return " ORDER BY " + strings.Join(parts, ", ") } func (c *pgCollection) documentExpression() string { parts := make([]string, 0, len(c.spec.allFields())*2) for _, field := range c.spec.columnFields() { parts = append(parts, fmt.Sprintf("'%s'", field), relationalIdent(relationalColumnName(field))) } relationFields := make([]string, 0, len(c.spec.ListRelations)) for field := range c.spec.ListRelations { relationFields = append(relationFields, field) } sort.Strings(relationFields) for _, field := range relationFields { relation := c.spec.ListRelations[field] expr := fmt.Sprintf( "(SELECT jsonb_agg(%s ORDER BY %s) FROM %s rel WHERE rel.%s = %s.%s)", "rel."+relationalIdent(relation.TargetColumn), "rel."+relationalIdent(relation.TargetColumn), relationalIdent(relation.Table), relationalIdent(relation.OwnerColumn), relationalIdent(c.spec.Table), relationalIdent("id"), ) parts = append(parts, fmt.Sprintf("'%s'", field), expr) } return "jsonb_strip_nulls(jsonb_build_object(" + strings.Join(parts, ", ") + "))" } func (c *pgCollection) syncListRelations(ctx context.Context, exec interface { Exec(context.Context, string, ...any) (pgconn.CommandTag, error) }, doc Doc) error { ownerID := strings.TrimSpace(documentString(doc["id"])) if ownerID == "" { return errors.New("relation owner id is required") } fields := make([]string, 0, len(c.spec.ListRelations)) for field := range c.spec.ListRelations { fields = append(fields, field) } sort.Strings(fields) for _, field := range fields { relation := c.spec.ListRelations[field] if _, err := exec.Exec( ctx, "DELETE FROM "+relationalIdent(relation.Table)+" WHERE "+relationalIdent(relation.OwnerColumn)+" = $1", ownerID, ); err != nil { return err } raw, exists := doc[field] if !exists || raw == nil { continue } values, ok := asAnySlice(raw) if !ok { return fmt.Errorf("%s.%s must be an array", c.spec.Name, field) } for _, value := range values { target := strings.TrimSpace(documentString(value)) if target == "" { continue } query := fmt.Sprintf( "INSERT INTO %s (%s, %s) VALUES ($1, $2) ON CONFLICT DO NOTHING", relationalIdent(relation.Table), relationalIdent(relation.OwnerColumn), relationalIdent(relation.TargetColumn), ) if _, err := exec.Exec(ctx, query, ownerID, target); err != nil { return err } } } return nil } func (c *pgCollection) listRelationFilterClause(relation relationalListRelation, raw any, args *[]any) (string, error) { base := "SELECT 1 FROM " + relationalIdent(relation.Table) + " rel WHERE rel." + relationalIdent(relation.OwnerColumn) + " = " + relationalIdent(c.spec.Table) + "." + relationalIdent("id") if operators, ok := asDocMap(raw); ok && containsRelationalOperator(operators) { parts := make([]string, 0, len(operators)) for operator, operand := range operators { switch operator { case "$exists": exists, ok := operand.(bool) if !ok { return "", errors.New("$exists requires a boolean") } if exists { parts = append(parts, "EXISTS ("+base+")") } else { parts = append(parts, "NOT EXISTS ("+base+")") } case "$in": values, ok := asAnySlice(operand) if !ok || len(values) == 0 { parts = append(parts, "FALSE") continue } matches := make([]string, 0, len(values)) for _, value := range values { *args = append(*args, documentString(value)) matches = append(matches, "rel."+relationalIdent(relation.TargetColumn)+fmt.Sprintf(" = $%d", len(*args))) } parts = append(parts, "EXISTS ("+base+" AND ("+strings.Join(matches, " OR ")+"))") case "$ne": *args = append(*args, documentString(operand)) parts = append(parts, "NOT EXISTS ("+base+" AND rel."+relationalIdent(relation.TargetColumn)+fmt.Sprintf(" = $%d)", len(*args))) default: return "", fmt.Errorf("unsupported relation operator %s", operator) } } return "(" + strings.Join(parts, " AND ") + ")", nil } *args = append(*args, documentString(raw)) return "EXISTS (" + base + " AND rel." + relationalIdent(relation.TargetColumn) + fmt.Sprintf(" = $%d)", len(*args)), nil } func relationalDBValue(spec relationalResourceSpec, field string, value any) any { if value == nil { return nil } if spec.isJSON(field) { raw, err := json.Marshal(value) if err != nil { return "null" } return string(raw) } if isBooleanResourceField(spec.Name, field) { return anyToBool(value) } if isNumericField(field) { number, ok := numericValue(value) if !ok { return int64(0) } if field == "score" || field == "progress" { return number } return int64(number) } switch typed := value.(type) { case time.Time: return typed.UTC().Format(time.RFC3339Nano) default: return documentString(value) } } func (r *pgSingleResult) Decode(target any) error { if r.err != nil { return r.err } return assignRelationalDocument(target, r.doc) } func (c *pgCursor) Close(context.Context) error { return nil } func (c *pgCursor) All(_ context.Context, target any) error { switch typed := target.(type) { case *[]Doc: out := make([]Doc, len(c.docs)) for i, doc := range c.docs { out[i] = cloneDoc(doc) } *typed = out return nil default: raw, err := json.Marshal(c.docs) if err != nil { return err } return json.Unmarshal(raw, target) } } func (c *pgCursor) Next(context.Context) bool { if c.index >= len(c.docs) { return false } c.index++ return true } func (c *pgCursor) Decode(target any) error { if c.index == 0 || c.index > len(c.docs) { return errors.New("cursor is not positioned on a row") } return assignRelationalDocument(target, c.docs[c.index-1]) } func assignRelationalDocument(target any, doc Doc) error { switch typed := target.(type) { case *Doc: *typed = cloneDoc(doc) return nil default: raw, err := json.Marshal(doc) if err != nil { return err } return json.Unmarshal(raw, target) } } func normalizeRelationalDocument(value any) (Doc, error) { switch typed := value.(type) { case Doc: return cloneDoc(typed), nil case map[string]any: return cloneDoc(Doc(typed)), nil default: raw, err := json.Marshal(value) if err != nil { return nil, err } var doc Doc if err := json.Unmarshal(raw, &doc); err != nil { return nil, err } return doc, nil } } func decodeRelationalDocument(raw []byte) (Doc, error) { var doc Doc if err := json.Unmarshal(raw, &doc); err != nil { return nil, err } return doc, nil } func cloneDoc(doc Doc) Doc { out := make(Doc, len(doc)) for key, value := range doc { out[key] = value } return out } func applyRelationalUpdate(doc Doc, update Doc) { if setRaw, ok := update["$set"]; ok { if set, ok := asDocMap(setRaw); ok { for key, value := range set { doc[key] = value } } } if unsetRaw, ok := update["$unset"]; ok { if unset, ok := asDocMap(unsetRaw); ok { for key := range unset { delete(doc, key) } } } } func containsRelationalOperator(value Doc) bool { for key := range value { if strings.HasPrefix(key, "$") { return true } } return false } func asDocMap(value any) (Doc, bool) { switch typed := value.(type) { case Doc: return typed, true case map[string]any: return Doc(typed), true default: return nil, false } } func asAnySlice(value any) ([]any, bool) { switch typed := value.(type) { case List: return []any(typed), true case []any: return typed, true case []string: out := make([]any, len(typed)) for i, value := range typed { out[i] = value } return out, true default: return nil, false } } func documentString(value any) string { if value == nil { return "" } if text, ok := value.(string); ok { return text } return fmt.Sprint(value) } func documentInt64(value any) int64 { number, ok := numericValue(value) if !ok { return 0 } return int64(number) }