fix(cloud): materialize chunk uploads as mutations

Author: Alan-TheGentleman

internal/cloud/cloudserver/mutations_test.go

@@ -12,6 +12,8 @@ import (
 	"time"
 
 	"github.com/Gentleman-Programming/engram/internal/cloud/cloudstore"
+	"github.com/Gentleman-Programming/engram/internal/store"
+	engramsync "github.com/Gentleman-Programming/engram/internal/sync"
 )
 
 // ─── Fakes for mutation tests ─────────────────────────────────────────────────
@@ -50,6 +52,125 @@ func (s *fakeMutationStore) IsProjectSyncEnabled(project string) (bool, error) {
 	return true, nil // default: enabled
 }
 
+func (s *fakeMutationStore) WriteChunk(ctx context.Context, project string, chunkID, createdBy, clientCreatedAt string, payload []byte) error {
+	if _, exists := s.chunks[chunkID]; exists {
+		return s.fakeStore.WriteChunk(ctx, project, chunkID, createdBy, clientCreatedAt, payload)
+	}
+	if err := s.fakeStore.WriteChunk(ctx, project, chunkID, createdBy, clientCreatedAt, payload); err != nil {
+		return err
+	}
+	var chunk engramsync.ChunkData
+	if err := json.Unmarshal(payload, &chunk); err != nil {
+		return err
+	}
+	batch := make([]MutationEntry, 0, len(chunk.Sessions)+len(chunk.Observations)+len(chunk.Prompts))
+	for _, session := range chunk.Sessions {
+		body, _ := json.Marshal(session)
+		batch = append(batch, MutationEntry{Project: project, Entity: store.SyncEntitySession, EntityKey: strings.TrimSpace(session.ID), Op: store.SyncOpUpsert, Payload: body})
+	}
+	for _, observation := range chunk.Observations {
+		body, _ := json.Marshal(observation)
+		batch = append(batch, MutationEntry{Project: project, Entity: store.SyncEntityObservation, EntityKey: strings.TrimSpace(observation.SyncID), Op: store.SyncOpUpsert, Payload: body})
+	}
+	for _, prompt := range chunk.Prompts {
+		body, _ := json.Marshal(prompt)
+		batch = append(batch, MutationEntry{Project: project, Entity: store.SyncEntityPrompt, EntityKey: strings.TrimSpace(prompt.SyncID), Op: store.SyncOpUpsert, Payload: body})
+	}
+	_, err := s.InsertMutationBatch(ctx, batch)
+	return err
+}
+
+func TestChunkPushMaterializesMutationsForAutosyncPull(t *testing.T) {
+	ms := newFakeMutationStore()
+	srv := newMutationTestServer(ms, "secret", []string{"proj-a"})
+	body := strings.NewReader(`{
+		"project":"proj-a",
+		"created_by":"tester",
+		"data":{
+			"sessions":[{"id":"s-1","directory":"/tmp/s-1","started_at":"2026-04-29T10:00:00Z"}],
+			"observations":[{"sync_id":"obs-1","session_id":"s-1","type":"decision","title":"Decision","content":"Content","scope":"project","created_at":"2026-04-29T10:01:00Z","updated_at":"2026-04-29T10:01:00Z"}],
+			"prompts":[{"sync_id":"prompt-1","session_id":"s-1","content":"Prompt","created_at":"2026-04-29T10:02:00Z"}]
+		}
+	}`)
+
+	pushRec := httptest.NewRecorder()
+	pushReq := httptest.NewRequest(http.MethodPost, "/sync/push", body)
+	pushReq.Header.Set("Authorization", "Bearer secret")
+	pushReq.Header.Set("Content-Type", "application/json")
+	srv.Handler().ServeHTTP(pushRec, pushReq)
+	if pushRec.Code != http.StatusOK {
+		t.Fatalf("expected chunk push 200, got %d body=%q", pushRec.Code, pushRec.Body.String())
+	}
+	if len(ms.chunks) != 1 {
+		t.Fatalf("expected one stored chunk, got %d", len(ms.chunks))
+	}
+	if len(ms.mutations) != 3 {
+		t.Fatalf("expected 3 materialized mutations, got %d: %+v", len(ms.mutations), ms.mutations)
+	}
+	if ms.mutations[0].Entity != store.SyncEntitySession || ms.mutations[1].Entity != store.SyncEntityObservation || ms.mutations[2].Entity != store.SyncEntityPrompt {
+		t.Fatalf("expected session/observation/prompt order, got %+v", ms.mutations)
+	}
+	if ms.mutations[1].Project != "proj-a" || ms.mutations[1].EntityKey != "obs-1" || ms.mutations[1].Op != store.SyncOpUpsert {
+		t.Fatalf("unexpected materialized observation mutation: %+v", ms.mutations[1])
+	}
+
+	pullRec := httptest.NewRecorder()
+	pullReq := httptest.NewRequest(http.MethodGet, "/sync/mutations/pull?since_seq=0&limit=100", nil)
+	pullReq.Header.Set("Authorization", "Bearer secret")
+	srv.Handler().ServeHTTP(pullRec, pullReq)
+	if pullRec.Code != http.StatusOK {
+		t.Fatalf("expected mutation pull 200, got %d body=%q", pullRec.Code, pullRec.Body.String())
+	}
+	var pulled struct {
+		Mutations []StoredMutation `json:"mutations"`
+	}
+	if err := json.NewDecoder(pullRec.Body).Decode(&pulled); err != nil {
+		t.Fatalf("decode pull response: %v", err)
+	}
+	if len(pulled.Mutations) != 3 || pulled.Mutations[1].Entity != store.SyncEntityObservation || pulled.Mutations[1].EntityKey != "obs-1" {
+		t.Fatalf("expected pulled observation mutation after chunk push, got %+v", pulled.Mutations)
+	}
+}
+
+func TestChunkPushReplayDoesNotDuplicateMaterializedMutations(t *testing.T) {
+	ms := newFakeMutationStore()
+	srv := newMutationTestServer(ms, "secret", []string{"proj-a"})
+	body := `{"project":"proj-a","created_by":"tester","data":{"sessions":[{"id":"s-1","directory":"/tmp/s-1"}],"observations":[{"sync_id":"obs-1","session_id":"s-1","type":"decision","title":"Decision","content":"Content","scope":"project"}]}}`
+
+	for i := 0; i < 2; i++ {
+		rec := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodPost, "/sync/push", strings.NewReader(body))
+		req.Header.Set("Authorization", "Bearer secret")
+		req.Header.Set("Content-Type", "application/json")
+		srv.Handler().ServeHTTP(rec, req)
+		if rec.Code != http.StatusOK {
+			t.Fatalf("push %d expected 200, got %d body=%q", i+1, rec.Code, rec.Body.String())
+		}
+	}
+	if len(ms.chunks) != 1 {
+		t.Fatalf("expected one stored chunk after replay, got %d", len(ms.chunks))
+	}
+	if len(ms.mutations) != 2 {
+		t.Fatalf("expected replay to keep 2 materialized mutations, got %d: %+v", len(ms.mutations), ms.mutations)
+	}
+}
+
+func TestMalformedChunkRejectsBeforeMaterialization(t *testing.T) {
+	ms := newFakeMutationStore()
+	srv := newMutationTestServer(ms, "secret", []string{"proj-a"})
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/sync/push", strings.NewReader(`{"project":"proj-a","created_by":"tester","data":{"observations":[{"sync_id":"obs-1","session_id":"missing","type":"decision","title":"Decision","content":"Content","scope":"project"}]}}`))
+	req.Header.Set("Authorization", "Bearer secret")
+	req.Header.Set("Content-Type", "application/json")
+	srv.Handler().ServeHTTP(rec, req)
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("expected malformed chunk 400, got %d body=%q", rec.Code, rec.Body.String())
+	}
+	if len(ms.chunks) != 0 || len(ms.mutations) != 0 {
+		t.Fatalf("expected no chunk or mutation after malformed push, chunks=%d mutations=%d", len(ms.chunks), len(ms.mutations))
+	}
+}
+
 func (s *fakeMutationStore) InsertMutationBatch(ctx context.Context, batch []MutationEntry) ([]int64, error) {
 	if s.errInsert != nil {
 		return nil, s.errInsert

internal/cloud/cloudstore/cloudstore.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/Gentleman-Programming/engram/internal/cloud"
 	"github.com/Gentleman-Programming/engram/internal/cloud/chunkcodec"
+	"github.com/Gentleman-Programming/engram/internal/store"
 	engramsync "github.com/Gentleman-Programming/engram/internal/sync"
 	"github.com/jackc/pgx/v5/pgconn"
 	_ "github.com/jackc/pgx/v5/stdlib"
@@ -230,8 +231,27 @@ func (cs *CloudStore) WriteChunk(ctx context.Context, project, chunkID, createdB
 		return fmt.Errorf("cloudstore: read existing chunk: %w", err)
 	}
 
+	chunk, err := parseChunkData(payload)
+	if err != nil {
+		return fmt.Errorf("cloudstore: parse chunk for materialization: %w", err)
+	}
+	mutations, err := materializedChunkMutations(project, chunk)
+	if err != nil {
+		return err
+	}
+
+	tx, err := cs.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("cloudstore: begin write chunk tx: %w", err)
+	}
+	defer func() {
+		if tx != nil {
+			_ = tx.Rollback()
+		}
+	}()
+
 	counts := summarizeChunk(payload)
-	_, err = cs.db.ExecContext(ctx, `
+	_, err = tx.ExecContext(ctx, `
 		INSERT INTO cloud_chunks (project_name, chunk_id, created_by, client_created_at, payload, sessions_count, observations_count, prompts_count)
 		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
 		project, strings.TrimSpace(chunkID), strings.TrimSpace(createdBy), originCreatedAt, payload, counts.sessions, counts.observations, counts.prompts)
@@ -246,9 +266,16 @@ func (cs *CloudStore) WriteChunk(ctx context.Context, project, chunkID, createdB
 		}
 		return fmt.Errorf("cloudstore: write chunk: %w", err)
 	}
-	if err := cs.indexChunkSessions(ctx, project, payload); err != nil {
+	if err := cs.indexChunkSessionsWith(ctx, tx, project, payload); err != nil {
+		return err
+	}
+	if err := insertMaterializedMutations(ctx, tx, mutations); err != nil {
 		return err
 	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("cloudstore: commit write chunk: %w", err)
+	}
+	tx = nil
 	cs.invalidateDashboardReadModel()
 	return nil
 }
@@ -296,12 +323,20 @@ func (cs *CloudStore) KnownSessionIDs(ctx context.Context, project string) (map[
 }
 
 func (cs *CloudStore) indexChunkSessions(ctx context.Context, project string, payload []byte) error {
+	return cs.indexChunkSessionsWith(ctx, cs.db, project, payload)
+}
+
+type chunkSessionIndexer interface {
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+}
+
+func (cs *CloudStore) indexChunkSessionsWith(ctx context.Context, execer chunkSessionIndexer, project string, payload []byte) error {
 	sessionIDs := collectSessionIDsFromPayload(payload)
 	if len(sessionIDs) == 0 {
 		return nil
 	}
 	for sessionID := range sessionIDs {
-		if _, err := cs.db.ExecContext(ctx,
+		if _, err := execer.ExecContext(ctx,
 			`INSERT INTO cloud_project_sessions (project_name, session_id) VALUES ($1, $2) ON CONFLICT (project_name, session_id) DO NOTHING`,
 			project, sessionID,
 		); err != nil {
@@ -311,6 +346,67 @@ func (cs *CloudStore) indexChunkSessions(ctx context.Context, project string, pa
 	return nil
 }
 
+func materializedChunkMutations(project string, chunk engramsync.ChunkData) ([]MutationEntry, error) {
+	project = strings.TrimSpace(project)
+	entries := make([]MutationEntry, 0, len(chunk.Sessions)+len(chunk.Observations)+len(chunk.Prompts))
+
+	for i, session := range chunk.Sessions {
+		entityKey := strings.TrimSpace(session.ID)
+		if entityKey == "" {
+			return nil, fmt.Errorf("cloudstore: materialize chunk: sessions[%d].id is required", i)
+		}
+		payload, err := json.Marshal(session)
+		if err != nil {
+			return nil, fmt.Errorf("cloudstore: materialize chunk session %q: %w", entityKey, err)
+		}
+		entries = append(entries, MutationEntry{Project: project, Entity: store.SyncEntitySession, EntityKey: entityKey, Op: store.SyncOpUpsert, Payload: payload})
+	}
+
+	for i, observation := range chunk.Observations {
+		entityKey := strings.TrimSpace(observation.SyncID)
+		if entityKey == "" {
+			return nil, fmt.Errorf("cloudstore: materialize chunk: observations[%d].sync_id is required", i)
+		}
+		payload, err := json.Marshal(observation)
+		if err != nil {
+			return nil, fmt.Errorf("cloudstore: materialize chunk observation %q: %w", entityKey, err)
+		}
+		entries = append(entries, MutationEntry{Project: project, Entity: store.SyncEntityObservation, EntityKey: entityKey, Op: store.SyncOpUpsert, Payload: payload})
+	}
+
+	for i, prompt := range chunk.Prompts {
+		entityKey := strings.TrimSpace(prompt.SyncID)
+		if entityKey == "" {
+			return nil, fmt.Errorf("cloudstore: materialize chunk: prompts[%d].sync_id is required", i)
+		}
+		payload, err := json.Marshal(prompt)
+		if err != nil {
+			return nil, fmt.Errorf("cloudstore: materialize chunk prompt %q: %w", entityKey, err)
+		}
+		entries = append(entries, MutationEntry{Project: project, Entity: store.SyncEntityPrompt, EntityKey: entityKey, Op: store.SyncOpUpsert, Payload: payload})
+	}
+
+	return entries, nil
+}
+
+func insertMaterializedMutations(ctx context.Context, tx *sql.Tx, entries []MutationEntry) error {
+	for _, entry := range entries {
+		payload := entry.Payload
+		if len(payload) == 0 {
+			payload = json.RawMessage("{}")
+		}
+		_, err := tx.ExecContext(ctx, `
+			INSERT INTO cloud_mutations (project, entity, entity_key, op, payload)
+			VALUES ($1, $2, $3, $4, $5)`,
+			strings.TrimSpace(entry.Project), strings.TrimSpace(entry.Entity), strings.TrimSpace(entry.EntityKey), strings.TrimSpace(entry.Op), payload,
+		)
+		if err != nil {
+			return fmt.Errorf("cloudstore: insert materialized chunk mutation %s/%s/%s: %w", entry.Project, entry.Entity, entry.EntityKey, err)
+		}
+	}
+	return nil
+}
+
 func (cs *CloudStore) backfillProjectSessionsFromChunks(ctx context.Context) error {
 	rows, err := cs.db.QueryContext(ctx, `SELECT project_name, payload FROM cloud_chunks`)
 	if err != nil {