feat: store layer with indexes, optimistic concurrency, GridFS, credential crypto (phase 2)
- mongo-driver v2 connection with bounded startup retry + /readyz gate - idempotent creation of all §4.9 indexes - UpdateVersioned: version-filtered updates, conflict vs not-found errors - AES-256-GCM Seal/Open for ticketing credentials, base64(nonce|ct) - GridFS file store: MIME sniffing, MAX_UPLOAD_MB cap, sha256 metadata - HMAC-signed short-lived file URL tokens (§5.1, 1h TTL) - integration tests against compose Mongo via docker-compose.test.yml overlay Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
etalon
@@ -0,0 +1,72 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
// EnsureIndexes creates every index from spec §4.9. CreateMany is idempotent:
|
||||
// existing identical indexes are no-ops, so this runs safely at every startup.
|
||||
func EnsureIndexes(ctx context.Context, db *mongo.Database) error {
|
||||
plan := map[string][]mongo.IndexModel{
|
||||
"users": {
|
||||
{Keys: bson.D{{Key: "email", Value: 1}}, Options: options.Index().SetUnique(true)},
|
||||
{Keys: bson.D{{Key: "auth.oidc.issuer", Value: 1}, {Key: "auth.oidc.subject", Value: 1}},
|
||||
Options: options.Index().SetUnique(true).SetSparse(true)},
|
||||
},
|
||||
"customers": {
|
||||
{Keys: bson.D{{Key: "name", Value: 1}}, Options: options.Index().SetUnique(true)},
|
||||
{Keys: bson.D{{Key: "consultantIds", Value: 1}}},
|
||||
},
|
||||
"pools": {
|
||||
{Keys: bson.D{{Key: "consultantId", Value: 1}, {Key: "developerId", Value: 1}},
|
||||
Options: options.Index().SetUnique(true)},
|
||||
{Keys: bson.D{{Key: "developerId", Value: 1}}},
|
||||
},
|
||||
"tasks": {
|
||||
{Keys: bson.D{{Key: "customerId", Value: 1}, {Key: "status", Value: 1}, {Key: "updatedAt", Value: -1}}},
|
||||
{Keys: bson.D{{Key: "consultantId", Value: 1}, {Key: "status", Value: 1}, {Key: "updatedAt", Value: -1}}},
|
||||
{Keys: bson.D{{Key: "assignee.userId", Value: 1}, {Key: "status", Value: 1}}},
|
||||
{Keys: bson.D{{Key: "rootId", Value: 1}}},
|
||||
{Keys: bson.D{{Key: "parentId", Value: 1}}},
|
||||
{Keys: bson.D{{Key: "external.system", Value: 1}, {Key: "external.key", Value: 1}, {Key: "customerId", Value: 1}},
|
||||
Options: options.Index().SetUnique(true).SetSparse(true)},
|
||||
{Keys: bson.D{{Key: "title", Value: "text"}, {Key: "description", Value: "text"}}},
|
||||
},
|
||||
"bountyAwards": {
|
||||
{Keys: bson.D{{Key: "developerId", Value: 1}, {Key: "awardedAt", Value: -1}}},
|
||||
{Keys: bson.D{{Key: "consultantId", Value: 1}, {Key: "awardedAt", Value: -1}}},
|
||||
{Keys: bson.D{{Key: "customerId", Value: 1}, {Key: "awardedAt", Value: -1}}},
|
||||
},
|
||||
"conversations": {
|
||||
{Keys: bson.D{{Key: "participantIds", Value: 1}, {Key: "lastMessageAt", Value: -1}}},
|
||||
{Keys: bson.D{{Key: "kind", Value: 1}, {Key: "customerId", Value: 1}}},
|
||||
},
|
||||
"messages": {
|
||||
// ULID _id is time-ordered, so this serves history pagination.
|
||||
{Keys: bson.D{{Key: "conversationId", Value: 1}, {Key: "_id", Value: 1}}},
|
||||
},
|
||||
"notifications": {
|
||||
{Keys: bson.D{{Key: "userId", Value: 1}, {Key: "readAt", Value: 1}, {Key: "createdAt", Value: -1}}},
|
||||
},
|
||||
"sessions": {
|
||||
{Keys: bson.D{{Key: "expiresAt", Value: 1}}, Options: options.Index().SetExpireAfterSeconds(0)},
|
||||
{Keys: bson.D{{Key: "userId", Value: 1}}},
|
||||
},
|
||||
"auditLog": {
|
||||
{Keys: bson.D{{Key: "at", Value: -1}}},
|
||||
{Keys: bson.D{{Key: "entityId", Value: 1}, {Key: "at", Value: -1}}},
|
||||
},
|
||||
}
|
||||
|
||||
for coll, models := range plan {
|
||||
if _, err := db.Collection(coll).Indexes().CreateMany(ctx, models); err != nil {
|
||||
return fmt.Errorf("create indexes for %s: %w", coll, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
)
|
||||
|
||||
// Optimistic-concurrency errors. HTTP handlers translate ErrVersionConflict
|
||||
// into a 409 with a "conflict" envelope (and the QoL reload toast).
|
||||
var (
|
||||
ErrNotFound = errors.New("document not found")
|
||||
ErrVersionConflict = errors.New("version conflict")
|
||||
)
|
||||
|
||||
// UpdateVersioned applies update to the document only if its version still
|
||||
// matches, bumping version and updatedAt atomically. update uses normal
|
||||
// operators ($set/$push/...); the helper merges its bookkeeping into them.
|
||||
func UpdateVersioned(ctx context.Context, coll *mongo.Collection, id string, version int64, update bson.M) error {
|
||||
merged := bson.M{}
|
||||
for op, fields := range update {
|
||||
merged[op] = fields
|
||||
}
|
||||
set, _ := merged["$set"].(bson.M)
|
||||
if set == nil {
|
||||
set = bson.M{}
|
||||
}
|
||||
set["updatedAt"] = time.Now().UTC()
|
||||
merged["$set"] = set
|
||||
inc, _ := merged["$inc"].(bson.M)
|
||||
if inc == nil {
|
||||
inc = bson.M{}
|
||||
}
|
||||
if _, clash := inc["version"]; clash {
|
||||
return fmt.Errorf("update must not touch version itself")
|
||||
}
|
||||
inc["version"] = int64(1)
|
||||
merged["$inc"] = inc
|
||||
|
||||
res, err := coll.UpdateOne(ctx, bson.M{"_id": id, "version": version}, merged)
|
||||
if err != nil {
|
||||
return fmt.Errorf("versioned update %s/%s: %w", coll.Name(), id, err)
|
||||
}
|
||||
if res.MatchedCount == 1 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Distinguish "gone" from "someone got there first".
|
||||
n, err := coll.CountDocuments(ctx, bson.M{"_id": id})
|
||||
if err != nil {
|
||||
return fmt.Errorf("versioned update %s/%s: existence check: %w", coll.Name(), id, err)
|
||||
}
|
||||
if n == 0 {
|
||||
return fmt.Errorf("%s/%s: %w", coll.Name(), id, ErrNotFound)
|
||||
}
|
||||
return fmt.Errorf("%s/%s at version %d: %w", coll.Name(), id, version, ErrVersionConflict)
|
||||
}
|
||||
|
||||
// InsertOne inserts a document map, stamping createdAt/updatedAt/version so
|
||||
// every collection follows the §4 conventions.
|
||||
func InsertOne(ctx context.Context, coll *mongo.Collection, id string, doc bson.M) error {
|
||||
now := time.Now().UTC()
|
||||
doc["_id"] = id
|
||||
doc["createdAt"] = now
|
||||
doc["updatedAt"] = now
|
||||
doc["version"] = int64(1)
|
||||
if _, err := coll.InsertOne(ctx, doc); err != nil {
|
||||
return fmt.Errorf("insert %s/%s: %w", coll.Name(), id, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
// Package store owns all MongoDB access: connection lifecycle, index
|
||||
// bootstrap, optimistic-concurrency updates, and (via internal/files) GridFS.
|
||||
// No other package talks to Mongo directly.
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/readpref"
|
||||
|
||||
"bountyboard/internal/config"
|
||||
)
|
||||
|
||||
type Store struct {
|
||||
Client *mongo.Client
|
||||
DB *mongo.Database
|
||||
}
|
||||
|
||||
// Connect builds the client (the driver dials lazily) and verifies
|
||||
// connectivity with a bounded retry so a compose race at boot does not kill
|
||||
// the app, then ensures all indexes.
|
||||
func Connect(ctx context.Context, cfg *config.Config, log *slog.Logger) (*Store, error) {
|
||||
opts := options.Client().
|
||||
ApplyURI(cfg.MongoURI).
|
||||
SetAppName("bountyboard").
|
||||
SetServerSelectionTimeout(5 * time.Second)
|
||||
if cfg.MongoUsername != "" {
|
||||
opts.SetAuth(options.Credential{Username: cfg.MongoUsername, Password: cfg.MongoPassword})
|
||||
}
|
||||
client, err := mongo.Connect(opts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("mongo connect: %w", err)
|
||||
}
|
||||
s := &Store{Client: client, DB: client.Database(cfg.MongoDB)}
|
||||
|
||||
const attempts = 6
|
||||
for i := 1; ; i++ {
|
||||
err = s.Ping(ctx)
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
if i == attempts || ctx.Err() != nil {
|
||||
return nil, fmt.Errorf("mongo ping (%d attempts): %w", i, err)
|
||||
}
|
||||
log.Warn("mongo not ready, retrying", "attempt", i, "err", err)
|
||||
select {
|
||||
case <-time.After(time.Duration(i) * time.Second):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
if err := EnsureIndexes(ctx, s.DB); err != nil {
|
||||
return nil, fmt.Errorf("ensure indexes: %w", err)
|
||||
}
|
||||
log.Info("mongo connected", "db", cfg.MongoDB)
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Store) Ping(ctx context.Context) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
defer cancel()
|
||||
return s.Client.Ping(ctx, readpref.Primary())
|
||||
}
|
||||
|
||||
func (s *Store) Close(ctx context.Context) error {
|
||||
return s.Client.Disconnect(ctx)
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
//go:build integration
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
|
||||
"bountyboard/internal/testutil"
|
||||
"bountyboard/internal/ulid"
|
||||
)
|
||||
|
||||
func TestEnsureIndexesIsIdempotent(t *testing.T) {
|
||||
db := testutil.MongoDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if err := EnsureIndexes(ctx, db); err != nil {
|
||||
t.Fatalf("first run: %v", err)
|
||||
}
|
||||
if err := EnsureIndexes(ctx, db); err != nil {
|
||||
t.Fatalf("second run must be a no-op: %v", err)
|
||||
}
|
||||
|
||||
// Spot-check a few critical indexes actually exist.
|
||||
checks := map[string]string{
|
||||
"users": "email_1",
|
||||
"tasks": "external.system_1_external.key_1_customerId_1",
|
||||
"sessions": "expiresAt_1",
|
||||
}
|
||||
for coll, wantName := range checks {
|
||||
cur, err := db.Collection(coll).Indexes().List(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var specs []bson.M
|
||||
if err := cur.All(ctx, &specs); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
found := false
|
||||
for _, s := range specs {
|
||||
if s["name"] == wantName {
|
||||
found = true
|
||||
// the unique upsert key must really be unique+sparse
|
||||
if coll == "tasks" {
|
||||
if s["unique"] != true || s["sparse"] != true {
|
||||
t.Errorf("%s/%s: want unique sparse, got %v", coll, wantName, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("%s: index %s not found in %v", coll, wantName, specs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateVersioned(t *testing.T) {
|
||||
db := testutil.MongoDB(t)
|
||||
ctx := context.Background()
|
||||
coll := db.Collection("tasks")
|
||||
|
||||
id := ulid.New()
|
||||
if err := InsertOne(ctx, coll, id, bson.M{"title": "original", "status": "imported"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Happy path bumps version and updatedAt.
|
||||
if err := UpdateVersioned(ctx, coll, id, 1, bson.M{"$set": bson.M{"title": "edited"}}); err != nil {
|
||||
t.Fatalf("update at correct version: %v", err)
|
||||
}
|
||||
var doc struct {
|
||||
Title string `bson:"title"`
|
||||
Version int64 `bson:"version"`
|
||||
UpdatedAt time.Time `bson:"updatedAt"`
|
||||
CreatedAt time.Time `bson:"createdAt"`
|
||||
}
|
||||
if err := coll.FindOne(ctx, bson.M{"_id": id}).Decode(&doc); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if doc.Title != "edited" || doc.Version != 2 {
|
||||
t.Fatalf("after update: title=%q version=%d", doc.Title, doc.Version)
|
||||
}
|
||||
if !doc.UpdatedAt.After(doc.CreatedAt) {
|
||||
t.Error("updatedAt must advance")
|
||||
}
|
||||
|
||||
// Stale version loses.
|
||||
err := UpdateVersioned(ctx, coll, id, 1, bson.M{"$set": bson.M{"title": "stale write"}})
|
||||
if !errors.Is(err, ErrVersionConflict) {
|
||||
t.Fatalf("want ErrVersionConflict, got %v", err)
|
||||
}
|
||||
|
||||
// Missing document is distinguished from a conflict.
|
||||
err = UpdateVersioned(ctx, coll, ulid.New(), 1, bson.M{"$set": bson.M{"title": "x"}})
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("want ErrNotFound, got %v", err)
|
||||
}
|
||||
|
||||
// Non-$set operators pass through; version still bumps.
|
||||
if err := UpdateVersioned(ctx, coll, id, 2, bson.M{
|
||||
"$push": bson.M{"timeline": bson.M{"event": "published"}},
|
||||
}); err != nil {
|
||||
t.Fatalf("push update: %v", err)
|
||||
}
|
||||
var doc2 struct {
|
||||
Version int64 `bson:"version"`
|
||||
Timeline []bson.M `bson:"timeline"`
|
||||
}
|
||||
if err := coll.FindOne(ctx, bson.M{"_id": id}).Decode(&doc2); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if doc2.Version != 3 || len(doc2.Timeline) != 1 {
|
||||
t.Fatalf("after push: version=%d timeline=%d", doc2.Version, len(doc2.Timeline))
|
||||
}
|
||||
|
||||
// Touching version inside the update is a programming error.
|
||||
if err := UpdateVersioned(ctx, coll, id, 3, bson.M{"$inc": bson.M{"version": 5}}); err == nil {
|
||||
t.Fatal("must reject updates that touch version")
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateVersionedConcurrentWriters(t *testing.T) {
|
||||
db := testutil.MongoDB(t)
|
||||
ctx := context.Background()
|
||||
coll := db.Collection("tasks")
|
||||
|
||||
id := ulid.New()
|
||||
if err := InsertOne(ctx, coll, id, bson.M{"n": 0}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Two writers race at the same version: exactly one wins.
|
||||
results := make(chan error, 2)
|
||||
for i := 0; i < 2; i++ {
|
||||
go func(i int) {
|
||||
results <- UpdateVersioned(ctx, coll, id, 1, bson.M{"$set": bson.M{"n": i + 1}})
|
||||
}(i)
|
||||
}
|
||||
var conflicts, oks int
|
||||
for i := 0; i < 2; i++ {
|
||||
switch err := <-results; {
|
||||
case err == nil:
|
||||
oks++
|
||||
case errors.Is(err, ErrVersionConflict):
|
||||
conflicts++
|
||||
default:
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
if oks != 1 || conflicts != 1 {
|
||||
t.Fatalf("oks=%d conflicts=%d, want exactly one of each", oks, conflicts)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user