Saved
", "status": "draft"}) + if w.Code != 201 { + t.Fatal(w.Code) + } + w, _ = doJSON(t, api, http.MethodPut, base+"/1", map[string]any{"expected_version": p["Version"], "title": "Updated"}) + if w.Code != 200 { + t.Fatal(w.Code) + } + w, h := historyResponse(t, api, base+"/history?limit=1") + if w.Code != 200 || w.Header().Get("Cache-Control") != "private, no-store" || len(h.Entries) != 1 || !h.HasMore || h.CurrentRevision != 2 || h.Entries[0].Actor != "user:authenticated-owner" || h.Entries[0].After[0].Content.BodyHTML != "Saved
" { + t.Fatal("protected saved history incorrect") + } + w, h = historyResponse(t, api, base+"/history?after_revision=1&limit=1") + if w.Code != 200 || len(h.Entries) != 1 || h.Entries[0].Before[0].Content.Title != "Rock" || h.Entries[0].After[0].Content.Title != "Updated" { + t.Fatal("saved update readback incorrect") + } + // Neither JSON nor an arbitrary actor header can supply trusted identity. + w, _ = doJSON(t, api, http.MethodPut, base+"/1", map[string]any{"expected_version": 2, "title": "Forged", "actor": "superadmin"}) + if w.Code != 400 { + t.Fatal("JSON actor accepted") + } + api.AuditActor = nil + r := httptest.NewRequest(http.MethodPut, base+"/1", strings.NewReader("{\"expected_version\":2,\"title\":\"Header ignored\"}")) + r.Header.Set("X-Actor", "superadmin") + w = httptest.NewRecorder() + api.ServeHTTP(w, r) + if w.Code != 200 { + t.Fatal(w.Code) + } + w, h = historyResponse(t, api, base+"/history?after_revision=2") + if w.Code != 200 || len(h.Entries) != 1 || h.Entries[0].Actor != "unattributed" { + t.Fatal("untrusted header fabricated identity") + } + w, _ = doJSON(t, api, http.MethodDelete, base+"/1", map[string]any{"expected_version": 3}) + if w.Code != 204 { + t.Fatal(w.Code) + } + w, h = historyResponse(t, api, base+"/history?after_revision=3") + if w.Code != 200 || len(h.Entries) != 1 || h.Entries[0].Operation != "delete" || h.Entries[0].Before[0].Content.Title != "Header ignored" { + t.Fatal("deleted page history missing") + } +} + +func TestAdminHistoryFailsClosedAndCanonicalRoutes(t *testing.T) { + path := "/api/v1/admin/tenants/234/pages/history" + for _, missing := range []string{"request", "tenant"} { + api := historyAPI() + if missing == "request" { + api.RequestAccess = nil + } else { + api.TenantAccess = nil + } + w, _ := historyResponse(t, api, path) + if w.Code != 503 { + t.Fatal("unconfigured history authority allowed") + } + } + api := historyAPI() + api.RequestAccess = func(*http.Request) bool { return false } + if w, _ := historyResponse(t, api, path); w.Code != 403 { + t.Fatal("denied request allowed") + } + api = historyAPI() + if w, _ := historyResponse(t, api, "/api/v1/admin/tenants/235/pages/history"); w.Code != 403 { + t.Fatal("foreign tenant history allowed") + } + for _, query := range []string{"?limit=0", "?limit=101", "?limit=01", "?after_revision=-1", "?after_revision=01", "?limit=1&limit=2", "?actor=x", "?limit=1;bad", "?limit=%zz"} { + if w, _ := historyResponse(t, api, path+query); w.Code != 400 { + t.Fatalf("noncanonical query allowed: %s", query) + } + } + for _, bad := range []string{"/api/v1/admin/tenants/0234/pages/history", path + "/1", path + "/history"} { + if w, _ := historyResponse(t, api, bad); w.Code != 404 { + t.Fatal("noncanonical history route allowed") + } + } +} diff --git a/store/batch_memory.go b/store/batch_memory.go index 7486816..b954294 100644 --- a/store/batch_memory.go +++ b/store/batch_memory.go @@ -24,23 +24,23 @@ func (s *MemoryPageStore) ReadPageState(_ context.Context, tenantID int64) (Page return PageSet{Scope: s.scopeLocked(tenantID), Revision: s.revisions[tenantID], Pages: s.statesLocked(tenantID)}, nil } -func (s *MemoryPageStore) ApplyPageBatch(_ context.Context, tenantID int64, batch PageBatch) (PageBatchReceipt, error) { +func (s *MemoryPageStore) ApplyPageBatch(ctx context.Context, tenantID int64, batch PageBatch) (PageBatchReceipt, error) { s.mu.Lock() defer s.mu.Unlock() - return s.applyBatchLocked(tenantID, batch, false) + return s.applyBatchLocked(ctx, tenantID, batch, false) } -func (s *MemoryPageStore) RollbackPageBatch(_ context.Context, tenantID int64, receipt PageBatchReceipt) (PageBatchReceipt, error) { +func (s *MemoryPageStore) RollbackPageBatch(ctx context.Context, tenantID int64, receipt PageBatchReceipt) (PageBatchReceipt, error) { batch, err := RollbackBatch(receipt) if err != nil { return PageBatchReceipt{}, err } s.mu.Lock() defer s.mu.Unlock() - return s.applyBatchLocked(tenantID, batch, true) + return s.applyBatchLocked(ctx, tenantID, batch, true) } -func (s *MemoryPageStore) applyBatchLocked(tenantID int64, batch PageBatch, restore bool) (PageBatchReceipt, error) { +func (s *MemoryPageStore) applyBatchLocked(ctx context.Context, tenantID int64, batch PageBatch, restore bool) (PageBatchReceipt, error) { if tenantID <= 0 { return PageBatchReceipt{}, ErrBatchInvalid } @@ -81,6 +81,12 @@ func (s *MemoryPageStore) applyBatchLocked(tenantID int64, batch PageBatch, rest result.After = s.statesLocked(tenantID) s.revisions[tenantID]++ result.AfterRevision = s.revisions[tenantID] + old, saved := ChangedPageStates(result.Before, result.After) + operation := "batch.apply" + if restore { + operation = "batch.rollback" + } + s.recordHistoryLocked(ctx, tenantID, operation, old, saved) result.Seal() return result, nil } diff --git a/store/history.go b/store/history.go new file mode 100644 index 0000000..b389f76 --- /dev/null +++ b/store/history.go @@ -0,0 +1,170 @@ +package store + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "strings" + "time" + "unicode" +) + +var ErrHistoryUnavailable = errors.New("page history unavailable; no write committed") +var ErrHistoryQuery = errors.New("invalid page history query") + +type pageWriteActorKey struct{} + +// WithPageWriteActor carries the host's authenticated stable identity into a +// store transaction. It grants no authority. Never populate it from a request +// body, arbitrary header, or unverified token claim. +func WithPageWriteActor(ctx context.Context, actor string) context.Context { + actor = strings.TrimSpace(actor) + if actor == "" || len(actor) > 256 || strings.ContainsFunc(actor, unicode.IsControl) { + actor = "unattributed" + } + return context.WithValue(ctx, pageWriteActorKey{}, actor) +} + +func pageWriteActor(ctx context.Context) string { + if actor, ok := ctx.Value(pageWriteActorKey{}).(string); ok { + return actor + } + return "unattributed" +} + +// PageHistoryEntry records only pages touched by one committed write. Deleted +// pages remain in Before. Versions, blocks, schedules and templates are saved +// content; Scope and Revision identify the durable tenant content namespace. +// Digest detects corruption; it is not a signature or proof of authorization. +type PageHistoryEntry struct { + Scope string `json:"scope"` + Revision int64 `json:"revision"` + CommittedAt time.Time `json:"committed_at"` + Actor string `json:"actor"` + Operation string `json:"operation"` + Before []PageState `json:"before"` + After []PageState `json:"after"` + Digest string `json:"digest"` +} + +func (e PageHistoryEntry) ContentDigest() string { + // Bind semantic content digests so JSONB key ordering does not break reads. + v := struct { + Scope string `json:"scope"` + Revision int64 `json:"revision"` + CommittedAt time.Time `json:"committed_at"` + Actor string `json:"actor"` + Operation string `json:"operation"` + Before []PageBaseline `json:"before"` + After []PageBaseline `json:"after"` + }{e.Scope, e.Revision, e.CommittedAt.UTC(), e.Actor, e.Operation, Baseline(e.Before), Baseline(e.After)} + b, _ := json.Marshal(v) + h := sha256.Sum256(b) + return hex.EncodeToString(h[:]) +} + +func NewPageHistoryEntry(ctx context.Context, scope string, revision int64, operation string, before, after []PageState, now time.Time) PageHistoryEntry { + e := PageHistoryEntry{scope, revision, now.UTC().Truncate(time.Microsecond), pageWriteActor(ctx), operation, cloneStates(before), cloneStates(after), ""} + e.Digest = e.ContentDigest() + return e +} + +func cloneStates(states []PageState) []PageState { + out := make([]PageState, 0, len(states)) + for _, p := range states { + out = append(out, PageState{p.ID, p.Version, ContentOf(p.Content.Page(1, p.ID, p.Version))}) + } + return out +} + +// ChangedPageStates includes version-only saves and excludes untouched pages. +func ChangedPageStates(before, after []PageState) ([]PageState, []PageState) { + b, a := map[int64]PageState{}, map[int64]PageState{} + for _, p := range before { + b[p.ID] = p + } + for _, p := range after { + a[p.ID] = p + } + changed := func(p PageState, other map[int64]PageState) bool { + o, ok := other[p.ID] + return !ok || p.Version != o.Version || p.Content.Digest() != o.Content.Digest() + } + var old, saved []PageState + for _, p := range before { + if changed(p, a) { + old = append(old, p) + } + } + for _, p := range after { + if changed(p, b) { + saved = append(saved, p) + } + } + return cloneStates(old), cloneStates(saved) +} + +type PageHistoryQuery struct { + AfterRevision int64 + Limit int +} + +func (q PageHistoryQuery) Validate() error { + if q.AfterRevision < 0 || q.Limit < 1 || q.Limit > 100 { + return ErrHistoryQuery + } + return nil +} + +// HistoryStartRevision is the head at adoption. No entries before it are +// claimed or fabricated. CurrentRevision is read atomically with this page of +// history. NextRevision is an exclusive cursor; HasMore signals another page. +type PageHistory struct { + Scope string `json:"scope"` + HistoryStartRevision int64 `json:"history_start_revision"` + CurrentRevision int64 `json:"current_revision"` + Entries []PageHistoryEntry `json:"entries"` + NextRevision int64 `json:"next_revision"` + HasMore bool `json:"has_more"` +} + +type PageHistoryStore interface { + ReadPageHistory(context.Context, int64, PageHistoryQuery) (PageHistory, error) +} + +func (s *MemoryPageStore) recordHistoryLocked(ctx context.Context, tenantID int64, operation string, before, after []PageState) { + s.history[tenantID] = append(s.history[tenantID], NewPageHistoryEntry(ctx, s.scopeLocked(tenantID), s.revisions[tenantID], operation, before, after, time.Now())) +} + +func (s *MemoryPageStore) ReadPageHistory(_ context.Context, tenantID int64, q PageHistoryQuery) (PageHistory, error) { + if tenantID <= 0 { + return PageHistory{}, ErrNotFound + } + if err := q.Validate(); err != nil { + return PageHistory{}, err + } + s.mu.RLock() + defer s.mu.RUnlock() + if s.scopes[tenantID] == "" { + return PageHistory{}, ErrNotFound + } + h := PageHistory{Scope: s.scopes[tenantID], CurrentRevision: s.revisions[tenantID], Entries: []PageHistoryEntry{}, NextRevision: q.AfterRevision} + for _, entry := range s.history[tenantID] { + if entry.Revision <= q.AfterRevision { + continue + } + if len(h.Entries) == q.Limit { + h.HasMore = true + break + } + e := entry + e.Before, e.After = cloneStates(e.Before), cloneStates(e.After) + h.Entries = append(h.Entries, e) + h.NextRevision = e.Revision + } + return h, nil +} + +var _ PageHistoryStore = (*MemoryPageStore)(nil) diff --git a/store/history_test.go b/store/history_test.go new file mode 100644 index 0000000..d0d96a8 --- /dev/null +++ b/store/history_test.go @@ -0,0 +1,38 @@ +package store_test + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/GoCodeAlone/workflow-plugin-cms/store" + "github.com/GoCodeAlone/workflow-plugin-cms/store/storetest" +) + +func TestMemorySavedPageHistory(t *testing.T) { + storetest.PageHistory(t, store.NewMemoryPageStore(), 1, 2) +} + +func TestHistoryActorAndSemanticDigest(t *testing.T) { + for _, actor := range []string{"", "\n", "user\x00injected", strings.Repeat("x", 257)} { + e := store.NewPageHistoryEntry(store.WithPageWriteActor(context.Background(), actor), "scope", 1, "create", nil, nil, time.Now()) + if e.Actor != "unattributed" { + t.Fatal("invalid actor accepted") + } + } + e := store.NewPageHistoryEntry(context.Background(), "scope", 1, "create", nil, nil, time.Now()) + if e.Actor != "unattributed" { + t.Fatal("missing identity fabricated") + } + changed := e + changed.Actor = "other" + if e.ContentDigest() == changed.ContentDigest() { + t.Fatal("actor not bound by digest") + } + changed = e + changed.Revision++ + if e.ContentDigest() == changed.ContentDigest() { + t.Fatal("revision not bound by digest") + } +} diff --git a/store/page.go b/store/page.go index e3b0e3e..6a42c20 100644 --- a/store/page.go +++ b/store/page.go @@ -118,11 +118,12 @@ type MemoryPageStore struct { pages map[int64]*Page // by ID revisions map[int64]int64 scopes map[int64]string + history map[int64][]PageHistoryEntry } // NewMemoryPageStore returns an empty in-memory store. func NewMemoryPageStore() *MemoryPageStore { - return &MemoryPageStore{pages: map[int64]*Page{}, revisions: map[int64]int64{}, scopes: map[int64]string{}} + return &MemoryPageStore{pages: map[int64]*Page{}, revisions: map[int64]int64{}, scopes: map[int64]string{}, history: map[int64][]PageHistoryEntry{}} } func (s *MemoryPageStore) scopeLocked(tenantID int64) string { @@ -134,7 +135,7 @@ func (s *MemoryPageStore) scopeLocked(tenantID int64) string { // Create inserts the page. Returns ErrPathConflict if (tenant, subsite, // path) is already taken. -func (s *MemoryPageStore) Create(_ context.Context, tenantID int64, p *Page) error { +func (s *MemoryPageStore) Create(ctx context.Context, tenantID int64, p *Page) error { if p == nil { return errors.New("page: nil") } @@ -163,6 +164,7 @@ func (s *MemoryPageStore) Create(_ context.Context, tenantID int64, p *Page) err s.pages[p.ID] = &cp s.scopeLocked(tenantID) s.revisions[tenantID]++ + s.recordHistoryLocked(ctx, tenantID, "create", nil, States([]*Page{&cp})) return nil } @@ -198,7 +200,7 @@ func (s *MemoryPageStore) GetByPath(_ context.Context, tenantID int64, subsite, // Update writes the page. Returns ErrNotFound if no record matches // (tenant_id, id). Bumps Version + UpdatedAt. -func (s *MemoryPageStore) Update(_ context.Context, tenantID int64, p *Page) error { +func (s *MemoryPageStore) Update(ctx context.Context, tenantID int64, p *Page) error { if p == nil { return errors.New("page: nil") } @@ -229,11 +231,12 @@ func (s *MemoryPageStore) Update(_ context.Context, tenantID int64, p *Page) err s.pages[p.ID] = &cp s.scopeLocked(tenantID) s.revisions[tenantID]++ + s.recordHistoryLocked(ctx, tenantID, "update", States([]*Page{existing}), States([]*Page{&cp})) return nil } // Delete removes the page. ErrNotFound on miss / wrong tenant. -func (s *MemoryPageStore) Delete(_ context.Context, tenantID, id int64, expectedVersion int) error { +func (s *MemoryPageStore) Delete(ctx context.Context, tenantID, id int64, expectedVersion int) error { s.mu.Lock() defer s.mu.Unlock() existing, ok := s.pages[id] @@ -246,6 +249,7 @@ func (s *MemoryPageStore) Delete(_ context.Context, tenantID, id int64, expected delete(s.pages, id) s.scopeLocked(tenantID) s.revisions[tenantID]++ + s.recordHistoryLocked(ctx, tenantID, "delete", States([]*Page{existing}), nil) return nil } diff --git a/store/postgres/batch.go b/store/postgres/batch.go index 05f4e83..600a517 100644 --- a/store/postgres/batch.go +++ b/store/postgres/batch.go @@ -195,6 +195,14 @@ func (s *Store) applyBatch(ctx context.Context, tenantID int64, batch store.Page if err != nil { return store.PageBatchReceipt{}, err } + old, saved := store.ChangedPageStates(result.Before, result.After) + operation := "batch.apply" + if restore { + operation = "batch.rollback" + } + if err := appendPageHistory(ctx, tx, tenantID, result.AfterRevision, operation, old, saved); err != nil { + return store.PageBatchReceipt{}, err + } if err = tx.Commit(ctx); err != nil { return store.PageBatchReceipt{}, store.ErrBatchInvalid } diff --git a/store/postgres/batch_integration_test.go b/store/postgres/batch_integration_test.go index aa13f1d..b88abe7 100644 --- a/store/postgres/batch_integration_test.go +++ b/store/postgres/batch_integration_test.go @@ -17,6 +17,10 @@ import ( // Uses a uniquely named schema and drops only that schema. Never uses the // application's search_path/tables. Connection values are not printed. func isolatedPostgres(t *testing.T) *Store { + return isolatedPostgresHistory(t, true) +} + +func isolatedPostgresHistory(t *testing.T, includeHistory bool) *Store { t.Helper() uri := os.Getenv("CMS_TEST_DATABASE_URL") if uri == "" { @@ -60,6 +64,15 @@ func isolatedPostgres(t *testing.T) *Store { if _, err = pool.Exec(ctx, string(migration)); err != nil { t.Fatal("isolated revision migration failed") } + if includeHistory { + historyMigration, err := os.ReadFile("migrations/0002_page_saved_history.up.sql") + if err != nil { + t.Fatal("required history test migration unavailable") + } + if _, err = pool.Exec(ctx, string(historyMigration)); err != nil { + t.Fatal("isolated history migration failed") + } + } return NewWithPool(pool) } diff --git a/store/postgres/history.go b/store/postgres/history.go new file mode 100644 index 0000000..d2dd627 --- /dev/null +++ b/store/postgres/history.go @@ -0,0 +1,112 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/GoCodeAlone/workflow-plugin-cms/store" + "github.com/jackc/pgx/v5" +) + +func pageBefore(ctx context.Context, tx pgx.Tx, tenantID, pageID int64) ([]store.PageState, error) { + p := &store.Page{} + err := scanPage(tx.QueryRow(ctx, pageSelectSQL()+` WHERE tenant_id=$1 AND id=$2 FOR UPDATE`, tenantID, pageID), p) + if errors.Is(err, pgx.ErrNoRows) { + return nil, store.ErrNotFound + } + if err != nil { + return nil, err + } + return store.States([]*store.Page{p}), nil +} + +func appendPageHistory(ctx context.Context, tx pgx.Tx, tenantID, revision int64, operation string, before, after []store.PageState) error { + var scope string + if err := tx.QueryRow(ctx, `SELECT content_scope FROM cms_page_revisions WHERE tenant_id=$1`, tenantID).Scan(&scope); err != nil { + return store.ErrHistoryUnavailable + } + e := store.NewPageHistoryEntry(ctx, scope, revision, operation, before, after, time.Now()) + old, err := json.Marshal(e.Before) + if err != nil { + return store.ErrHistoryUnavailable + } + saved, err := json.Marshal(e.After) + if err != nil { + return store.ErrHistoryUnavailable + } + // Refuse gaps caused by mixed old/new cooperative writers. Adoption starts + // at the explicit marker; later appends require the previous saved event. + tag, err := tx.Exec(ctx, `INSERT INTO cms_page_history (tenant_id,content_scope,revision,committed_at,actor,operation,before_pages,after_pages,digest) + SELECT tenant_id,content_scope,$2,$3,$4,$5,$6,$7,$8 FROM cms_page_revisions + WHERE tenant_id=$1 AND revision=$2 AND + (history_start_revision=$2-1 OR EXISTS(SELECT 1 FROM cms_page_history h WHERE h.tenant_id=$1 AND h.content_scope=cms_page_revisions.content_scope AND h.revision=$2-1))`, tenantID, revision, e.CommittedAt, e.Actor, e.Operation, old, saved, e.Digest) + if err != nil || tag.RowsAffected() != 1 { + return store.ErrHistoryUnavailable + } + return nil +} + +func (s *Store) ReadPageHistory(ctx context.Context, tenantID int64, q store.PageHistoryQuery) (store.PageHistory, error) { + if tenantID <= 0 { + return store.PageHistory{}, store.ErrNotFound + } + if err := q.Validate(); err != nil { + return store.PageHistory{}, err + } + // A read-only consistent snapshot does not take the page writer lock. + tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}) + if err != nil { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + defer tx.Rollback(ctx) + h := store.PageHistory{Entries: []store.PageHistoryEntry{}, NextRevision: q.AfterRevision} + err = tx.QueryRow(ctx, `SELECT content_scope,history_start_revision,revision FROM cms_page_revisions WHERE tenant_id=$1`, tenantID).Scan(&h.Scope, &h.HistoryStartRevision, &h.CurrentRevision) + if errors.Is(err, pgx.ErrNoRows) { + return store.PageHistory{}, store.ErrNotFound + } + if err != nil || h.HistoryStartRevision > h.CurrentRevision { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + var head int64 + if err := tx.QueryRow(ctx, `SELECT COALESCE(MAX(revision),$2) FROM cms_page_history WHERE tenant_id=$1 AND content_scope=$3`, tenantID, h.HistoryStartRevision, h.Scope).Scan(&head); err != nil || head != h.CurrentRevision { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + rows, err := tx.Query(ctx, `SELECT content_scope,revision,committed_at,actor,operation,before_pages,after_pages,digest FROM cms_page_history WHERE tenant_id=$1 AND revision>$2 ORDER BY revision LIMIT $3`, tenantID, q.AfterRevision, q.Limit+1) + if err != nil { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + defer rows.Close() + previous := q.AfterRevision + if previous < h.HistoryStartRevision { + previous = h.HistoryStartRevision + } + for rows.Next() { + var e store.PageHistoryEntry + var before, after []byte + if err := rows.Scan(&e.Scope, &e.Revision, &e.CommittedAt, &e.Actor, &e.Operation, &before, &after, &e.Digest); err != nil || json.Unmarshal(before, &e.Before) != nil || json.Unmarshal(after, &e.After) != nil { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + if e.Scope != h.Scope || e.Revision != previous+1 || e.Digest != e.ContentDigest() { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + previous = e.Revision + if len(h.Entries) == q.Limit { + h.HasMore = true + break + } + h.Entries = append(h.Entries, e) + h.NextRevision = e.Revision + } + if rows.Err() != nil { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + rows.Close() + if err := tx.Commit(ctx); err != nil { + return store.PageHistory{}, store.ErrHistoryUnavailable + } + return h, nil +} + +var _ store.PageHistoryStore = (*Store)(nil) diff --git a/store/postgres/history_integration_test.go b/store/postgres/history_integration_test.go new file mode 100644 index 0000000..925d1aa --- /dev/null +++ b/store/postgres/history_integration_test.go @@ -0,0 +1,246 @@ +package postgres + +import ( + "context" + "errors" + "os" + "reflect" + "sync" + "testing" + + "github.com/GoCodeAlone/workflow-plugin-cms/store" + "github.com/GoCodeAlone/workflow-plugin-cms/store/storetest" + "github.com/jackc/pgx/v5/pgxpool" +) + +func historyTenant(t *testing.T, s *Store, slug string) int64 { + t.Helper() + tenant := &store.Tenant{Slug: slug} + if err := s.CreateTenant(context.Background(), tenant); err != nil { + t.Fatal("isolated tenant failed") + } + return tenant.ID +} + +func TestPostgresSavedPageHistory(t *testing.T) { + s := isolatedPostgres(t) + tenantID := historyTenant(t, s, "saved-history") + otherID := historyTenant(t, s, "other-history") + storetest.PageHistory(t, s, tenantID, otherID) + before, err := s.ReadPageHistory(context.Background(), tenantID, store.PageHistoryQuery{Limit: 100}) + if err != nil { + t.Fatal(err) + } + // Reconnect to the same task-owned schema: deleted snapshots and actor + // history must survive loss of all in-memory store state. + pool, err := pgxpool.NewWithConfig(context.Background(), s.pool.Config()) + if err != nil { + t.Fatal("fresh isolated pool failed") + } + defer pool.Close() + after, err := NewWithPool(pool).ReadPageHistory(context.Background(), tenantID, store.PageHistoryQuery{Limit: 100}) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("saved history lost after reconnect") + } + for _, query := range []string{ + "UPDATE cms_page_history SET actor='rewritten'", + "DELETE FROM cms_page_history", + "TRUNCATE cms_page_history", + } { + if _, err := s.pool.Exec(context.Background(), query); err == nil { + t.Fatal("append-only history rewritable") + } + } +} + +func TestPostgresHistoryAdoptionDoesNotInventPast(t *testing.T) { + s := isolatedPostgresHistory(t, false) + ctx := context.Background() + tenantID := historyTenant(t, s, "existing-before-history") + // Simulate the previous released store before history is installed. + var pageID int64 + if err := s.pool.QueryRow(ctx, "INSERT INTO pages(tenant_id,path,title,status) VALUES($1,'/existing','Existing','draft') RETURNING id", tenantID).Scan(&pageID); err != nil { + t.Fatal("prehistory fixture failed") + } + if _, err := s.pool.Exec(ctx, "UPDATE cms_page_revisions SET revision=7 WHERE tenant_id=$1", tenantID); err != nil { + t.Fatal("prehistory ledger failed") + } + migration, err := os.ReadFile("migrations/0002_page_saved_history.up.sql") + if err != nil { + t.Fatal(err) + } + if _, err := s.pool.Exec(ctx, string(migration)); err != nil { + t.Fatal("isolated history adoption failed") + } + h, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}) + if err != nil || h.HistoryStartRevision != 7 || h.CurrentRevision != 7 || len(h.Entries) != 0 { + t.Fatal("historical actors or snapshots fabricated") + } + p, err := s.Get(ctx, tenantID, pageID) + if err != nil { + t.Fatal(err) + } + p.Title = "First recorded save" + if err := s.Update(store.WithPageWriteActor(ctx, "user:first-recorded"), tenantID, p); err != nil { + t.Fatal(err) + } + h, err = s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}) + if err != nil || len(h.Entries) != 1 || h.Entries[0].Revision != 8 || h.Entries[0].Before[0].Content.Title != "Existing" || h.Entries[0].Actor != "user:first-recorded" { + t.Fatal("adoption marker or first saved event incorrect") + } +} + +func TestPostgresHistoryFailureRollsBackEveryWrite(t *testing.T) { + s := isolatedPostgres(t) + ctx := context.Background() + tenantID := historyTenant(t, s, "history-failure") + p := &store.Page{Path: "/", Title: "Before", Status: store.StatusDraft} + if err := s.Create(ctx, tenantID, p); err != nil { + t.Fatal(err) + } + state, err := s.ReadPageState(ctx, tenantID) + if err != nil { + t.Fatal(err) + } + content := store.PageContent{Path: "/batch", Title: "Batch", Status: store.StatusDraft} + receipt, err := s.ApplyPageBatch(ctx, tenantID, store.BatchFor(state, []store.PageMutation{{Key: "created", Kind: "create", Content: &content}})) + if err != nil { + t.Fatal(err) + } + before, err := s.ReadPageState(ctx, tenantID) + if err != nil { + t.Fatal(err) + } + if _, err := s.pool.Exec(ctx, "DROP TABLE cms_page_history"); err != nil { + t.Fatal("isolated missing-history fixture failed") + } + p.Title = "Must not save" + if err := s.Update(ctx, tenantID, p); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("update bypassed missing history") + } + if err := s.Delete(ctx, tenantID, p.ID, 1); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("delete bypassed missing history") + } + if err := s.Create(ctx, tenantID, &store.Page{Path: "/new", Title: "Must not save", Status: store.StatusDraft}); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("create bypassed missing history") + } + content.Path = "/must-not-save" + if _, err := s.ApplyPageBatch(ctx, tenantID, store.BatchFor(before, []store.PageMutation{{Key: "refused", Kind: "create", Content: &content}})); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("batch bypassed missing history") + } + if _, err := s.RollbackPageBatch(ctx, tenantID, receipt); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("rollback bypassed missing history") + } + after, err := s.ReadPageState(ctx, tenantID) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("history failure partly committed page or revision") + } + if _, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("missing history reported as empty complete history") + } +} + +func TestPostgresMixedWriterGapRefuses(t *testing.T) { + s := isolatedPostgres(t) + ctx := context.Background() + tenantID := historyTenant(t, s, "mixed-writer") + p := &store.Page{Path: "/", Title: "Before", Status: store.StatusDraft} + if err := s.Create(ctx, tenantID, p); err != nil { + t.Fatal(err) + } + // A superseded writer advanced only the prior revision ledger. + if _, err := s.pool.Exec(ctx, "UPDATE cms_page_revisions SET revision=revision+1 WHERE tenant_id=$1", tenantID); err != nil { + t.Fatal("isolated mixed-writer fixture failed") + } + p.Title = "Must refuse" + if err := s.Update(ctx, tenantID, p); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("unrecorded revision gap accepted") + } + got, err := s.Get(ctx, tenantID, p.ID) + if err != nil || got.Title != "Before" || got.Version != 1 { + t.Fatal("gap refusal committed content") + } + if _, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("gap concealed by history reader") + } +} + +func TestPostgresConcurrentSaveRecordsOneWinner(t *testing.T) { + s := isolatedPostgres(t) + ctx := context.Background() + tenantID := historyTenant(t, s, "concurrent-history") + p := &store.Page{Path: "/", Title: "Loaded", Status: store.StatusDraft} + if err := s.Create(ctx, tenantID, p); err != nil { + t.Fatal(err) + } + ready := make(chan struct{}) + results := make(chan error, 2) + var group sync.WaitGroup + for _, title := range []string{"Owner save", "Agent save"} { + loaded, err := s.Get(ctx, tenantID, p.ID) + if err != nil { + t.Fatal(err) + } + loaded.Title = title + group.Add(1) + go func(page *store.Page) { + defer group.Done() + <-ready + results <- s.Update(store.WithPageWriteActor(ctx, page.Title), tenantID, page) + }(loaded) + } + close(ready) + group.Wait() + close(results) + saved, refused := 0, 0 + for err := range results { + if err == nil { + saved++ + } else if errors.Is(err, store.ErrVersionConflict) { + refused++ + } else { + t.Fatal(err) + } + } + h, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}) + got, getErr := s.Get(ctx, tenantID, p.ID) + if saved != 1 || refused != 1 || err != nil || getErr != nil || len(h.Entries) != 2 || h.CurrentRevision != 2 || h.Entries[1].After[0].Content.Title != got.Title || h.Entries[1].Actor != got.Title { + t.Fatal("concurrent save overwrote winner or recorded a refused write") + } +} + +func TestPostgresHistoryInsertFailureIsAtomic(t *testing.T) { + s := isolatedPostgres(t) + ctx := context.Background() + tenantID := historyTenant(t, s, "insert-history-failure") + p := &store.Page{Path: "/", Title: "Before", Status: store.StatusDraft} + if err := s.Create(ctx, tenantID, p); err != nil { + t.Fatal(err) + } + before, err := s.ReadPageState(ctx, tenantID) + if err != nil { + t.Fatal(err) + } + historyBefore, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}) + if err != nil { + t.Fatal(err) + } + // Inject a late failure at the actual history INSERT boundary after all + // batch page writes and the tenant revision increment have executed. + _, err = s.pool.Exec(ctx, "CREATE FUNCTION refuse_history_insert() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture refusal'; END $$; CREATE TRIGGER refuse_history_insert BEFORE INSERT ON cms_page_history FOR EACH ROW EXECUTE FUNCTION refuse_history_insert();") + if err != nil { + t.Fatal("isolated history failure trigger failed") + } + update := store.ContentOf(p) + update.Title = "Must roll back" + create := store.PageContent{Path: "/new", Title: "Must also roll back", Status: store.StatusDraft} + batch := store.BatchFor(before, []store.PageMutation{{Key: "update", Kind: "update", TargetID: p.ID, ExpectedVersion: p.Version, Content: &update}, {Key: "create", Kind: "create", Content: &create}}) + if _, err := s.ApplyPageBatch(ctx, tenantID, batch); !errors.Is(err, store.ErrHistoryUnavailable) { + t.Fatal("late history failure not refused") + } + after, err := s.ReadPageState(ctx, tenantID) + historyAfter, historyErr := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 20}) + if err != nil || historyErr != nil || !reflect.DeepEqual(before, after) || !reflect.DeepEqual(historyBefore, historyAfter) { + t.Fatal("late history failure partially committed") + } +} diff --git a/store/postgres/migrations/0002_page_saved_history.up.sql b/store/postgres/migrations/0002_page_saved_history.up.sql new file mode 100644 index 0000000..c3a19d7 --- /dev/null +++ b/store/postgres/migrations/0002_page_saved_history.up.sql @@ -0,0 +1,27 @@ +-- Explicit additional host prerequisite, after 0001 and under a write fence. +-- No fabricated backfill: history begins after each current tenant revision. +BEGIN; +ALTER TABLE cms_page_revisions ADD COLUMN history_start_revision BIGINT NOT NULL DEFAULT 0 CHECK (history_start_revision >= 0 AND history_start_revision <= revision); +UPDATE cms_page_revisions SET history_start_revision = revision; +CREATE TABLE cms_page_history ( + tenant_id BIGINT NOT NULL REFERENCES cms_page_revisions(tenant_id), + content_scope TEXT NOT NULL, + revision BIGINT NOT NULL CHECK (revision > 0), + committed_at TIMESTAMPTZ NOT NULL, + actor TEXT NOT NULL CHECK (length(actor) BETWEEN 1 AND 256), + operation TEXT NOT NULL CHECK (operation IN ('create','update','delete','batch.apply','batch.rollback')), + before_pages JSONB NOT NULL CHECK (jsonb_typeof(before_pages) = 'array'), + after_pages JSONB NOT NULL CHECK (jsonb_typeof(after_pages) = 'array'), + digest TEXT NOT NULL CHECK (digest ~ '^[0-9a-f]{64}$'), + PRIMARY KEY (tenant_id, revision) +); +CREATE FUNCTION cms_refuse_history_rewrite() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'saved page history is append-only'; +END; +$$; +CREATE TRIGGER cms_refuse_history_rewrite BEFORE UPDATE OR DELETE ON cms_page_history + FOR EACH ROW EXECUTE FUNCTION cms_refuse_history_rewrite(); +CREATE TRIGGER cms_refuse_history_truncate BEFORE TRUNCATE ON cms_page_history + FOR EACH STATEMENT EXECUTE FUNCTION cms_refuse_history_rewrite(); +COMMIT; diff --git a/store/postgres/store.go b/store/postgres/store.go index 4ceaed0..4d25840 100644 --- a/store/postgres/store.go +++ b/store/postgres/store.go @@ -245,7 +245,11 @@ func (s *Store) Create(ctx context.Context, tenantID int64, p *store.Page) error if err = createPage(ctx, tx, tenantID, p); err != nil { return err } - if _, err = advanceRevision(ctx, tx, tenantID); err != nil { + revision, err := advanceRevision(ctx, tx, tenantID) + if err != nil { + return err + } + if err := appendPageHistory(ctx, tx, tenantID, revision, "create", nil, store.States([]*store.Page{p})); err != nil { return err } return tx.Commit(ctx) @@ -310,15 +314,26 @@ func (s *Store) GetByPath(ctx context.Context, tenantID int64, subsite, path str // Update uses the caller's loaded version, never a newly fetched version. func (s *Store) Update(ctx context.Context, tenantID int64, p *store.Page) error { + if p == nil { + return errors.New("page: nil") + } tx, err := s.pageTransaction(ctx, tenantID) if err != nil { return err } defer tx.Rollback(ctx) + before, err := pageBefore(ctx, tx, tenantID, p.ID) + if err != nil { + return err + } if err = updatePage(ctx, tx, tenantID, p); err != nil { return err } - if _, err = advanceRevision(ctx, tx, tenantID); err != nil { + revision, err := advanceRevision(ctx, tx, tenantID) + if err != nil { + return err + } + if err := appendPageHistory(ctx, tx, tenantID, revision, "update", before, store.States([]*store.Page{p})); err != nil { return err } return tx.Commit(ctx) @@ -363,10 +378,18 @@ func (s *Store) Delete(ctx context.Context, tenantID int64, id int64, expectedVe return err } defer tx.Rollback(ctx) + before, err := pageBefore(ctx, tx, tenantID, id) + if err != nil { + return err + } if err = deletePage(ctx, tx, tenantID, id, expectedVersion); err != nil { return err } - if _, err = advanceRevision(ctx, tx, tenantID); err != nil { + revision, err := advanceRevision(ctx, tx, tenantID) + if err != nil { + return err + } + if err := appendPageHistory(ctx, tx, tenantID, revision, "delete", before, nil); err != nil { return err } return tx.Commit(ctx) diff --git a/store/storetest/history.go b/store/storetest/history.go new file mode 100644 index 0000000..a97a606 --- /dev/null +++ b/store/storetest/history.go @@ -0,0 +1,126 @@ +package storetest + +import ( + "context" + "encoding/json" + "errors" + "reflect" + "testing" + "time" + + "github.com/GoCodeAlone/workflow-plugin-cms/store" +) + +type HistoryStore interface { + store.PageStore + store.PageBatchStore + store.PageHistoryStore +} + +// PageHistory exercises the same saved-content contract in memory and actual +// PostgreSQL. Only committed writes create history; stale saves never retry. +func PageHistory(t *testing.T, s HistoryStore, tenantID, otherTenantID int64) { + t.Helper() + ctx := store.WithPageWriteActor(context.Background(), "user:owner-1") + scheduled := time.Date(2030, 11, 3, 6, 30, 0, 0, time.UTC) + p := &store.Page{Path: "/history", Title: "Original", BodyHTML: "Original
", BodyBlocks: json.RawMessage("{\"z\":[],\"a\":\"blocks\"}"), TemplateID: "studio", Status: store.StatusScheduled, PublishAt: &scheduled} + if err := s.Create(ctx, tenantID, p); err != nil { + t.Fatal(err) + } + original := store.ContentOf(p) + q := store.PageHistoryQuery{Limit: 100} + h, err := s.ReadPageHistory(ctx, tenantID, q) + if err != nil || len(h.Entries) != 1 { + t.Fatalf("create history: %v %#v", err, h) + } + e := h.Entries[0] + if e.Actor != "user:owner-1" || e.Operation != "create" || e.Revision != 1 || h.CurrentRevision != 1 || h.HistoryStartRevision != 0 || e.Scope != h.Scope || len(e.Before) != 0 || len(e.After) != 1 || e.After[0].Content.Digest() != original.Digest() || e.Digest != e.ContentDigest() || e.CommittedAt.IsZero() { + t.Fatal("create did not bind saved content, trusted actor and revision") + } + // Returned values and caller pointers must not rewrite stored snapshots. + e.After[0].Content.BodyBlocks[0] = ' ' + e.After[0].Content.PublishAt = nil + p.BodyBlocks = nil + p.PublishAt = nil + p.Title = "Saved change" + if err := s.Update(store.WithPageWriteActor(ctx, "agent:review-1"), tenantID, p); err != nil { + t.Fatal(err) + } + saved := store.ContentOf(p) + stale := saved.Page(tenantID, p.ID, 1) + stale.Title = "Stale must not save" + if err := s.Update(ctx, tenantID, stale); !errors.Is(err, store.ErrVersionConflict) { + t.Fatal("stale save accepted") + } + if err := s.Delete(ctx, tenantID, p.ID, 1); !errors.Is(err, store.ErrVersionConflict) { + t.Fatal("stale delete accepted") + } + h, err = s.ReadPageHistory(ctx, tenantID, q) + if err != nil || len(h.Entries) != 2 || h.CurrentRevision != 2 || h.Entries[0].After[0].Content.Digest() != original.Digest() || h.Entries[1].Before[0].Content.Digest() != original.Digest() || h.Entries[1].After[0].Content.Digest() != saved.Digest() || h.Entries[1].Actor != "agent:review-1" || h.Entries[1].After[0].Version != 2 { + t.Fatal("update history changed earlier snapshot or recorded failed writes") + } + if err := s.Delete(context.Background(), tenantID, p.ID, p.Version); err != nil { + t.Fatal(err) + } + if _, err := s.Get(ctx, tenantID, p.ID); !errors.Is(err, store.ErrNotFound) { + t.Fatal("deleted page still present") + } + h, err = s.ReadPageHistory(ctx, tenantID, q) + if err != nil || len(h.Entries) != 3 || h.Entries[2].Operation != "delete" || h.Entries[2].Actor != "unattributed" || len(h.Entries[2].After) != 0 || h.Entries[2].Before[0].Content.Digest() != saved.Digest() { + t.Fatal("deleted saved content not retained") + } + if other, err := s.ReadPageHistory(ctx, otherTenantID, q); err != nil && !errors.Is(err, store.ErrNotFound) || len(other.Entries) != 0 { + t.Fatal("history leaked cross-tenant") + } + first, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{Limit: 2}) + if err != nil || len(first.Entries) != 2 || !first.HasMore || first.NextRevision != 2 { + t.Fatal("history first page incorrect") + } + last, err := s.ReadPageHistory(ctx, tenantID, store.PageHistoryQuery{AfterRevision: first.NextRevision, Limit: 2}) + if err != nil || len(last.Entries) != 1 || last.HasMore || last.NextRevision != 3 { + t.Fatal("history cursor replayed or omitted entries") + } + for _, query := range []store.PageHistoryQuery{{Limit: 0}, {Limit: 101}, {AfterRevision: -1, Limit: 1}} { + if _, err := s.ReadPageHistory(ctx, tenantID, query); !errors.Is(err, store.ErrHistoryQuery) { + t.Fatal("invalid history query accepted") + } + } + if _, err := s.ReadPageHistory(ctx, 0, q); !errors.Is(err, store.ErrNotFound) { + t.Fatal("unscoped history read accepted") + } + // A batch event retains only its changed pages and rollback gets a new + // tenant revision; receipt sealing and replay prevention stay authoritative. + keep := &store.Page{Path: "/keep", Title: "Untouched", Status: store.StatusDraft} + if err := s.Create(ctx, tenantID, keep); err != nil { + t.Fatal(err) + } + state, err := s.ReadPageState(ctx, tenantID) + if err != nil { + t.Fatal(err) + } + content := store.PageContent{Path: "/batch", Title: "Batch", Status: store.StatusDraft} + receipt, err := s.ApplyPageBatch(ctx, tenantID, store.BatchFor(state, []store.PageMutation{{Key: "create", Kind: "create", Content: &content}})) + if err != nil { + t.Fatal(err) + } + if _, err := s.RollbackPageBatch(ctx, tenantID, receipt); err != nil { + t.Fatal(err) + } + h, err = s.ReadPageHistory(ctx, tenantID, q) + if err != nil || len(h.Entries) != 6 || h.Entries[4].Operation != "batch.apply" || len(h.Entries[4].Before) != 0 || len(h.Entries[4].After) != 1 || h.Entries[5].Operation != "batch.rollback" || len(h.Entries[5].Before) != 1 || len(h.Entries[5].After) != 0 || h.Entries[4].After[0].ID == keep.ID { + t.Fatal("batch/rollback history omitted changes or included untouched pages") + } + before := h + if _, err := s.RollbackPageBatch(ctx, tenantID, receipt); !errors.Is(err, store.ErrVersionConflict) { + t.Fatal("rollback replay accepted") + } + after, err := s.ReadPageHistory(ctx, tenantID, q) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("failed rollback wrote history") + } + for _, entry := range after.Entries { + if entry.Digest != entry.ContentDigest() { + t.Fatal("history digest mismatch") + } + } +}