899 lines
23 KiB
Go
899 lines
23 KiB
Go
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)
|
|
}
|