From 8e2294e19d85afe96b4cfa0c7ccc0c24437f27d8 Mon Sep 17 00:00:00 2001 From: Algis Dumbris Date: Fri, 13 Mar 2026 12:12:27 +0200 Subject: [PATCH] feat: implement attachments with content-addressable storage Add file attachment support with SHA-256 content-addressable storage, automatic deduplication, MIME detection, and garbage collection for orphaned files. Includes MCP tools (upload_attachment, download_attachment, gc_attachments), REST API endpoints for Web UI, and comprehensive tests. Co-Authored-By: Claude Opus 4.6 --- cmd/synapbus/main.go | 18 +- internal/api/attachments_handler.go | 160 +++++++++++++ internal/api/router.go | 15 +- internal/api/traces_handler_test.go | 20 +- internal/attachments/cas.go | 145 ++++++++++++ internal/attachments/cas_test.go | 263 +++++++++++++++++++++ internal/attachments/doc.go | 12 + internal/attachments/mime.go | 120 ++++++++++ internal/attachments/mime_test.go | 162 +++++++++++++ internal/attachments/model.go | 62 +++++ internal/attachments/service.go | 224 ++++++++++++++++++ internal/attachments/service_test.go | 265 ++++++++++++++++++++++ internal/attachments/sqlite_store.go | 167 ++++++++++++++ internal/attachments/sqlite_store_test.go | 226 ++++++++++++++++++ internal/attachments/store.go | 25 ++ internal/mcp/server.go | 8 + internal/mcp/tools_attachments.go | 161 +++++++++++++ 17 files changed, 2039 insertions(+), 14 deletions(-) create mode 100644 internal/api/attachments_handler.go create mode 100644 internal/attachments/cas.go create mode 100644 internal/attachments/cas_test.go create mode 100644 internal/attachments/doc.go create mode 100644 internal/attachments/mime.go create mode 100644 internal/attachments/mime_test.go create mode 100644 internal/attachments/model.go create mode 100644 internal/attachments/service.go create mode 100644 internal/attachments/service_test.go create mode 100644 internal/attachments/sqlite_store.go create mode 100644 internal/attachments/sqlite_store_test.go create mode 100644 internal/attachments/store.go create mode 100644 internal/mcp/tools_attachments.go diff --git a/cmd/synapbus/main.go b/cmd/synapbus/main.go index 3f2a76c..70b8f64 100644 --- a/cmd/synapbus/main.go +++ b/cmd/synapbus/main.go @@ -9,6 +9,7 @@ import ( "net/http" "os" "os/signal" + "path/filepath" "strconv" "strings" "syscall" @@ -19,6 +20,7 @@ import ( "github.com/smart-mcp-proxy/synapbus/internal/agents" "github.com/smart-mcp-proxy/synapbus/internal/api" + "github.com/smart-mcp-proxy/synapbus/internal/attachments" "github.com/smart-mcp-proxy/synapbus/internal/auth" "github.com/smart-mcp-proxy/synapbus/internal/channels" mcpserver "github.com/smart-mcp-proxy/synapbus/internal/mcp" @@ -180,6 +182,16 @@ func runServe(cmd *cobra.Command, args []string) error { channelStore := channels.NewSQLiteChannelStore(db.DB) channelService := channels.NewService(channelStore, msgService, tracer) + // Create attachment service + attachmentsDir := filepath.Join(dataDir, "attachments") + cas, err := attachments.NewCAS(attachmentsDir, slog.Default()) + if err != nil { + return fmt.Errorf("create CAS engine: %w", err) + } + attachmentStore := attachments.NewSQLiteStore(db.DB, slog.Default()) + attachmentService := attachments.NewService(attachmentStore, cas, slog.Default()) + slog.Info("attachment service initialized", "dir", attachmentsDir) + // Initialize auth subsystem authSecret := make([]byte, 32) if _, err := rand.Read(authSecret); err != nil { @@ -221,7 +233,7 @@ func runServe(cmd *cobra.Command, args []string) error { } // Create MCP server - mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService) + mcpSrv := mcpserver.NewMCPServer(msgService, agentService, channelService, attachmentService) startTime := time.Now() // Set up chi router @@ -250,8 +262,8 @@ func runServe(cmd *cobra.Command, args []string) error { // MCP SSE endpoint r.Mount("/mcp", mcpSrv.SSEHandler()) - // Mount API routes (traces, export, stats, metrics) - apiRouter := api.NewRouter(traceStore, metrics) + // Mount API routes (traces, export, stats, metrics, attachments) + apiRouter := api.NewRouter(traceStore, metrics, attachmentService) r.Mount("/", apiRouter) // Start HTTP server diff --git a/internal/api/attachments_handler.go b/internal/api/attachments_handler.go new file mode 100644 index 0000000..34a2f06 --- /dev/null +++ b/internal/api/attachments_handler.go @@ -0,0 +1,160 @@ +package api + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + + "github.com/go-chi/chi/v5" + + "github.com/smart-mcp-proxy/synapbus/internal/attachments" +) + +// AttachmentsHandler provides REST API endpoints for attachment operations. +// These endpoints are intended for the Web UI, not for agent-to-agent use +// (agents use MCP tools instead). +type AttachmentsHandler struct { + service *attachments.Service + logger *slog.Logger +} + +// NewAttachmentsHandler creates a new attachments REST handler. +func NewAttachmentsHandler(service *attachments.Service) *AttachmentsHandler { + return &AttachmentsHandler{ + service: service, + logger: slog.Default().With("component", "api-attachments"), + } +} + +// Download streams an attachment file to the client. +// GET /api/attachments/{hash} +func (h *AttachmentsHandler) Download(w http.ResponseWriter, r *http.Request) { + hash := chi.URLParam(r, "hash") + if hash == "" { + http.Error(w, `{"error":"hash parameter required"}`, http.StatusBadRequest) + return + } + + result, err := h.service.Download(r.Context(), hash) + if err != nil { + switch err { + case attachments.ErrNotFound, attachments.ErrFileMissing: + http.Error(w, `{"error":"attachment not found"}`, http.StatusNotFound) + default: + h.logger.Error("download attachment failed", "hash", hash, "error", err) + http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError) + } + return + } + defer result.Content.Close() + + w.Header().Set("Content-Type", result.MIMEType) + if result.Size > 0 { + w.Header().Set("Content-Length", fmt.Sprintf("%d", result.Size)) + } + + // Images are displayed inline; everything else triggers a download. + if attachments.IsImageType(result.MIMEType) { + w.Header().Set("Content-Disposition", fmt.Sprintf("inline; filename=%q", result.Filename)) + } else { + w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", result.Filename)) + } + + if _, err := streamContent(w, result.Content); err != nil { + h.logger.Error("stream attachment failed", "hash", hash, "error", err) + } +} + +// Metadata returns attachment metadata as JSON. +// GET /api/attachments/{hash}/meta +func (h *AttachmentsHandler) Metadata(w http.ResponseWriter, r *http.Request) { + hash := chi.URLParam(r, "hash") + if hash == "" { + http.Error(w, `{"error":"hash parameter required"}`, http.StatusBadRequest) + return + } + + result, err := h.service.Download(r.Context(), hash) + if err != nil { + switch err { + case attachments.ErrNotFound, attachments.ErrFileMissing: + http.Error(w, `{"error":"attachment not found"}`, http.StatusNotFound) + default: + h.logger.Error("get attachment metadata failed", "hash", hash, "error", err) + http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError) + } + return + } + // Close the content reader immediately since we only need metadata. + result.Content.Close() + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "hash": result.Hash, + "original_filename": result.Filename, + "mime_type": result.MIMEType, + "size": result.Size, + "is_image": attachments.IsImageType(result.MIMEType), + }) +} + +// Upload handles multipart file uploads from the Web UI. +// POST /api/attachments +func (h *AttachmentsHandler) Upload(w http.ResponseWriter, r *http.Request) { + // Limit request body to MaxFileSize + overhead for multipart headers. + r.Body = http.MaxBytesReader(w, r.Body, attachments.MaxFileSize+1024*1024) + + if err := r.ParseMultipartForm(attachments.MaxFileSize); err != nil { + http.Error(w, `{"error":"file too large or invalid multipart form"}`, http.StatusBadRequest) + return + } + + file, header, err := r.FormFile("file") + if err != nil { + http.Error(w, `{"error":"file field required"}`, http.StatusBadRequest) + return + } + defer file.Close() + + // Extract uploader identity from context (set by auth middleware). + uploadedBy := "web-ui" + if ownerID, ok := OwnerIDFromContext(r.Context()); ok { + uploadedBy = fmt.Sprintf("owner-%d", ownerID) + } + + req := attachments.UploadRequest{ + Content: file, + Filename: header.Filename, + UploadedBy: uploadedBy, + } + + result, err := h.service.Upload(r.Context(), req) + if err != nil { + switch err { + case attachments.ErrEmptyFile: + http.Error(w, `{"error":"empty file not allowed"}`, http.StatusBadRequest) + case attachments.ErrFileTooLarge: + http.Error(w, `{"error":"file exceeds maximum size of 50MB"}`, http.StatusRequestEntityTooLarge) + default: + h.logger.Error("upload attachment failed", "error", err) + http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError) + } + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]any{ + "hash": result.Hash, + "size": result.Size, + "mime_type": result.MIMEType, + "original_filename": result.Filename, + }) +} + +// streamContent copies the reader to the response writer. +func streamContent(dst http.ResponseWriter, src io.Reader) (int64, error) { + return io.Copy(dst, src) +} diff --git a/internal/api/router.go b/internal/api/router.go index 27c0968..c2a33cc 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -5,12 +5,14 @@ import ( "github.com/go-chi/chi/v5" + "github.com/smart-mcp-proxy/synapbus/internal/attachments" "github.com/smart-mcp-proxy/synapbus/internal/trace" ) // NewRouter creates a chi router with all API routes configured. // metricsInstance may be nil if metrics are disabled. -func NewRouter(traceStore trace.TraceStore, metricsInstance *trace.Metrics) chi.Router { +// attachmentService may be nil if attachments are not configured. +func NewRouter(traceStore trace.TraceStore, metricsInstance *trace.Metrics, attachmentService *attachments.Service) chi.Router { r := chi.NewRouter() // Global middleware @@ -28,6 +30,17 @@ func NewRouter(traceStore trace.TraceStore, metricsInstance *trace.Metrics) chi. r.Get("/api/traces/stats", tracesHandler.TraceStats) }) + // Attachment API routes (for Web UI) + if attachmentService != nil { + attachmentsHandler := NewAttachmentsHandler(attachmentService) + r.Get("/api/attachments/{hash}", attachmentsHandler.Download) + r.Get("/api/attachments/{hash}/meta", attachmentsHandler.Metadata) + r.Group(func(r chi.Router) { + r.Use(OwnerAuthMiddleware) + r.Post("/api/attachments", attachmentsHandler.Upload) + }) + } + // Metrics endpoint (unauthenticated, only registered when enabled) if metricsInstance != nil { r.Get("/metrics", func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/api/traces_handler_test.go b/internal/api/traces_handler_test.go index 9432b65..702dfa2 100644 --- a/internal/api/traces_handler_test.go +++ b/internal/api/traces_handler_test.go @@ -81,7 +81,7 @@ func TestListTraces_OwnerIsolation(t *testing.T) { seedTraces(t, store, "1", "alice-bot", []string{"send_message", "read_inbox", "send_message"}) seedTraces(t, store, "2", "bob-bot", []string{"send_message", "error"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) t.Run("owner 1 sees only own traces", func(t *testing.T) { rr := makeRequest(t, router, "GET", "/api/traces", "1") @@ -141,7 +141,7 @@ func TestListTraces_Filters(t *testing.T) { seedTraces(t, store, "1", "agent-a", []string{"send_message", "read_inbox", "send_message", "error"}) seedTraces(t, store, "1", "agent-b", []string{"send_message", "join_channel"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) t.Run("filter by agent_name", func(t *testing.T) { rr := makeRequest(t, router, "GET", "/api/traces?agent_name=agent-a", "1") @@ -223,7 +223,7 @@ func TestListTraces_ResponseFormat(t *testing.T) { seedTraces(t, store, "1", "agent", []string{"send_message"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) rr := makeRequest(t, router, "GET", "/api/traces", "1") if rr.Code != http.StatusOK { t.Fatalf("status = %d", rr.Code) @@ -267,7 +267,7 @@ func TestTraceStats(t *testing.T) { seedTraces(t, store, "1", "agent", []string{"send_message", "send_message", "read_inbox", "error"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) rr := makeRequest(t, router, "GET", "/api/traces/stats", "1") if rr.Code != http.StatusOK { t.Fatalf("status = %d", rr.Code) @@ -288,7 +288,7 @@ func TestExportTraces_JSON(t *testing.T) { seedTraces(t, store, "1", "agent", []string{"send_message", "read_inbox"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) rr := makeRequest(t, router, "GET", "/api/traces/export?format=json", "1") if rr.Code != http.StatusOK { t.Fatalf("status = %d", rr.Code) @@ -318,7 +318,7 @@ func TestExportTraces_CSV(t *testing.T) { seedTraces(t, store, "1", "agent", []string{"send_message", "read_inbox"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) rr := makeRequest(t, router, "GET", "/api/traces/export?format=csv", "1") if rr.Code != http.StatusOK { t.Fatalf("status = %d", rr.Code) @@ -353,7 +353,7 @@ func TestExportTraces_Empty(t *testing.T) { db := newTestDB(t) store := trace.NewSQLiteTraceStore(db) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) t.Run("empty JSON export", func(t *testing.T) { rr := makeRequest(t, router, "GET", "/api/traces/export?format=json", "1") @@ -400,7 +400,7 @@ func TestMetricsEndpoint(t *testing.T) { metrics.IncError() metrics.SetActiveAgents(3) - router := NewRouter(store, metrics) + router := NewRouter(store, metrics, nil) rr := makeRequest(t, router, "GET", "/metrics", "") if rr.Code != http.StatusOK { t.Fatalf("status = %d", rr.Code) @@ -426,7 +426,7 @@ func TestMetricsEndpoint(t *testing.T) { }) t.Run("metrics disabled returns 404", func(t *testing.T) { - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) rr := makeRequest(t, router, "GET", "/metrics", "") if rr.Code != http.StatusNotFound { t.Errorf("status = %d, want %d", rr.Code, http.StatusNotFound) @@ -443,7 +443,7 @@ func TestMultiOwnerIsolation(t *testing.T) { seedTraces(t, store, "2", "bob-bot", []string{"send_message", "error", "join_channel"}) seedTraces(t, store, "3", "charlie-bot", []string{"send_message"}) - router := NewRouter(store, nil) + router := NewRouter(store, nil, nil) for _, tc := range []struct { ownerID string diff --git a/internal/attachments/cas.go b/internal/attachments/cas.go new file mode 100644 index 0000000..2c85ec9 --- /dev/null +++ b/internal/attachments/cas.go @@ -0,0 +1,145 @@ +package attachments + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "sync" +) + +// CAS implements content-addressable storage on the local filesystem. +// Files are stored at {baseDir}/{hash[0:2]}/{hash[2:4]}/{hash}. +type CAS struct { + baseDir string + logger *slog.Logger + mu sync.Mutex // protects concurrent writes of the same hash +} + +// NewCAS creates a new content-addressable store rooted at baseDir. +// The directory tree is created automatically if it does not exist. +func NewCAS(baseDir string, logger *slog.Logger) (*CAS, error) { + if err := os.MkdirAll(baseDir, 0o755); err != nil { + return nil, fmt.Errorf("create CAS base directory: %w", err) + } + return &CAS{ + baseDir: baseDir, + logger: logger.With("component", "cas"), + }, nil +} + +// shardPath returns the full filesystem path for the given hash. +func (c *CAS) shardPath(hash string) string { + return filepath.Join(c.baseDir, hash[0:2], hash[2:4], hash) +} + +// Write streams content from r, computing its SHA-256 hash while writing to +// a temporary file. On success the temp file is atomically renamed to the +// content-addressed path. If the file already exists (dedup), the temp file +// is removed and the existing hash is returned. +// +// Returns ErrEmptyFile if r yields zero bytes. +func (c *CAS) Write(r io.Reader) (hash string, size int64, err error) { + // Write to a temp file while computing the hash. + tmpFile, err := os.CreateTemp(c.baseDir, ".cas-upload-*") + if err != nil { + return "", 0, fmt.Errorf("create temp file: %w", err) + } + tmpPath := tmpFile.Name() + + // Ensure cleanup on any error path. + defer func() { + if err != nil { + tmpFile.Close() + os.Remove(tmpPath) + } + }() + + hasher := sha256.New() + w := io.MultiWriter(tmpFile, hasher) + + size, err = io.Copy(w, r) + if err != nil { + return "", 0, fmt.Errorf("write content: %w", err) + } + + if size == 0 { + err = ErrEmptyFile + return "", 0, err + } + + if err = tmpFile.Close(); err != nil { + return "", 0, fmt.Errorf("close temp file: %w", err) + } + + hash = hex.EncodeToString(hasher.Sum(nil)) + destPath := c.shardPath(hash) + + c.mu.Lock() + defer c.mu.Unlock() + + // Check for dedup: if the file already exists, skip the rename. + if _, statErr := os.Stat(destPath); statErr == nil { + os.Remove(tmpPath) + c.logger.Debug("dedup: file already exists", "hash", hash) + return hash, size, nil + } + + // Create shard directories. + destDir := filepath.Dir(destPath) + if err = os.MkdirAll(destDir, 0o755); err != nil { + return "", 0, fmt.Errorf("create shard directory: %w", err) + } + + // Atomic rename. + if err = os.Rename(tmpPath, destPath); err != nil { + return "", 0, fmt.Errorf("rename temp file: %w", err) + } + + c.logger.Info("file stored", "hash", hash, "size", size) + return hash, size, nil +} + +// Read opens the file identified by hash and returns a ReadCloser. +// Returns ErrNotFound if the file does not exist on disk. +func (c *CAS) Read(hash string) (io.ReadCloser, error) { + path := c.shardPath(hash) + f, err := os.Open(path) + if err != nil { + if os.IsNotExist(err) { + return nil, ErrNotFound + } + return nil, fmt.Errorf("open file: %w", err) + } + return f, nil +} + +// Exists returns true if the file identified by hash exists on disk. +func (c *CAS) Exists(hash string) bool { + _, err := os.Stat(c.shardPath(hash)) + return err == nil +} + +// Delete removes the file identified by hash from disk. +// Returns the size of the deleted file, or 0 if the file did not exist. +func (c *CAS) Delete(hash string) (int64, error) { + path := c.shardPath(hash) + info, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + return 0, nil + } + return 0, fmt.Errorf("stat file: %w", err) + } + + size := info.Size() + if err := os.Remove(path); err != nil { + return 0, fmt.Errorf("remove file: %w", err) + } + + c.logger.Info("file deleted", "hash", hash, "size", size) + return size, nil +} diff --git a/internal/attachments/cas_test.go b/internal/attachments/cas_test.go new file mode 100644 index 0000000..f300282 --- /dev/null +++ b/internal/attachments/cas_test.go @@ -0,0 +1,263 @@ +package attachments + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "io" + "log/slog" + "os" + "path/filepath" + "strings" + "sync" + "testing" +) + +func newTestCAS(t *testing.T) *CAS { + t.Helper() + dir := t.TempDir() + cas, err := NewCAS(filepath.Join(dir, "attachments"), slog.Default()) + if err != nil { + t.Fatalf("NewCAS: %v", err) + } + return cas +} + +func sha256Hex(data []byte) string { + h := sha256.Sum256(data) + return hex.EncodeToString(h[:]) +} + +func TestCAS_Write(t *testing.T) { + tests := []struct { + name string + content []byte + wantErr error + }{ + { + name: "normal file", + content: []byte("hello world"), + }, + { + name: "binary content", + content: []byte{0x00, 0x01, 0xff, 0xfe, 0x89, 0x50, 0x4e, 0x47}, + }, + { + name: "zero bytes", + content: []byte{}, + wantErr: ErrEmptyFile, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cas := newTestCAS(t) + hash, size, err := cas.Write(bytes.NewReader(tt.content)) + + if tt.wantErr != nil { + if err != tt.wantErr { + t.Fatalf("expected error %v, got %v", tt.wantErr, err) + } + return + } + if err != nil { + t.Fatalf("Write: %v", err) + } + + wantHash := sha256Hex(tt.content) + if hash != wantHash { + t.Errorf("hash = %s, want %s", hash, wantHash) + } + if size != int64(len(tt.content)) { + t.Errorf("size = %d, want %d", size, len(tt.content)) + } + + // Verify sharded path exists. + path := filepath.Join(cas.baseDir, hash[0:2], hash[2:4], hash) + if _, err := os.Stat(path); err != nil { + t.Errorf("file not found at sharded path: %v", err) + } + }) + } +} + +func TestCAS_Read(t *testing.T) { + cas := newTestCAS(t) + content := []byte("test content for reading") + + hash, _, err := cas.Write(bytes.NewReader(content)) + if err != nil { + t.Fatalf("Write: %v", err) + } + + t.Run("existing file", func(t *testing.T) { + rc, err := cas.Read(hash) + if err != nil { + t.Fatalf("Read: %v", err) + } + defer rc.Close() + + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !bytes.Equal(got, content) { + t.Errorf("content mismatch: got %q, want %q", got, content) + } + }) + + t.Run("missing file", func(t *testing.T) { + _, err := cas.Read("deadbeef" + strings.Repeat("0", 56)) + if err != ErrNotFound { + t.Errorf("expected ErrNotFound, got %v", err) + } + }) +} + +func TestCAS_Exists(t *testing.T) { + cas := newTestCAS(t) + content := []byte("existence check") + + hash, _, err := cas.Write(bytes.NewReader(content)) + if err != nil { + t.Fatalf("Write: %v", err) + } + + if !cas.Exists(hash) { + t.Error("Exists returned false for stored file") + } + if cas.Exists("deadbeef" + strings.Repeat("0", 56)) { + t.Error("Exists returned true for non-existent file") + } +} + +func TestCAS_Delete(t *testing.T) { + cas := newTestCAS(t) + content := []byte("deletable content") + + hash, _, err := cas.Write(bytes.NewReader(content)) + if err != nil { + t.Fatalf("Write: %v", err) + } + + t.Run("delete existing", func(t *testing.T) { + size, err := cas.Delete(hash) + if err != nil { + t.Fatalf("Delete: %v", err) + } + if size != int64(len(content)) { + t.Errorf("deleted size = %d, want %d", size, len(content)) + } + if cas.Exists(hash) { + t.Error("file still exists after delete") + } + }) + + t.Run("delete non-existent", func(t *testing.T) { + size, err := cas.Delete("deadbeef" + strings.Repeat("0", 56)) + if err != nil { + t.Fatalf("Delete non-existent: %v", err) + } + if size != 0 { + t.Errorf("deleted size = %d, want 0", size) + } + }) +} + +func TestCAS_Dedup(t *testing.T) { + cas := newTestCAS(t) + content := []byte("duplicate content") + + hash1, _, err := cas.Write(bytes.NewReader(content)) + if err != nil { + t.Fatalf("first Write: %v", err) + } + + hash2, _, err := cas.Write(bytes.NewReader(content)) + if err != nil { + t.Fatalf("second Write: %v", err) + } + + if hash1 != hash2 { + t.Errorf("hashes differ: %s vs %s", hash1, hash2) + } + + // Verify only one file on disk. + path := filepath.Join(cas.baseDir, hash1[0:2], hash1[2:4], hash1) + info, err := os.Stat(path) + if err != nil { + t.Fatalf("file not found: %v", err) + } + if info.Size() != int64(len(content)) { + t.Errorf("file size = %d, want %d", info.Size(), len(content)) + } +} + +func TestCAS_ConcurrentWrite(t *testing.T) { + cas := newTestCAS(t) + content := []byte("concurrent content") + expected := sha256Hex(content) + + const n = 10 + var wg sync.WaitGroup + errs := make([]error, n) + hashes := make([]string, n) + + wg.Add(n) + for i := 0; i < n; i++ { + go func(idx int) { + defer wg.Done() + h, _, err := cas.Write(bytes.NewReader(content)) + hashes[idx] = h + errs[idx] = err + }(i) + } + wg.Wait() + + for i := 0; i < n; i++ { + if errs[i] != nil { + t.Errorf("goroutine %d: %v", i, errs[i]) + } + if hashes[i] != expected { + t.Errorf("goroutine %d: hash = %s, want %s", i, hashes[i], expected) + } + } + + // Verify single file on disk. + if !cas.Exists(expected) { + t.Error("file does not exist after concurrent writes") + } +} + +func TestCAS_DirectoryStructure(t *testing.T) { + cas := newTestCAS(t) + content := []byte("structure check") + + hash, _, err := cas.Write(bytes.NewReader(content)) + if err != nil { + t.Fatalf("Write: %v", err) + } + + // Check two-level sharding. + level1 := filepath.Join(cas.baseDir, hash[0:2]) + level2 := filepath.Join(level1, hash[2:4]) + file := filepath.Join(level2, hash) + + for _, p := range []string{level1, level2} { + info, err := os.Stat(p) + if err != nil { + t.Fatalf("directory %s not found: %v", p, err) + } + if !info.IsDir() { + t.Errorf("%s is not a directory", p) + } + } + + info, err := os.Stat(file) + if err != nil { + t.Fatalf("file not found: %v", err) + } + if info.IsDir() { + t.Error("file is a directory") + } +} diff --git a/internal/attachments/doc.go b/internal/attachments/doc.go new file mode 100644 index 0000000..9117c6f --- /dev/null +++ b/internal/attachments/doc.go @@ -0,0 +1,12 @@ +// Package attachments provides content-addressable file storage for SynapBus. +// +// Files are stored on the local filesystem using a two-level sharded directory +// structure based on the SHA-256 hash of the content: +// +// {data_dir}/attachments/{hash[0:2]}/{hash[2:4]}/{hash} +// +// Deduplication is automatic: identical content produces the same hash and is +// stored only once on disk, while each upload creates its own metadata row in +// SQLite. Garbage collection removes orphaned files that are no longer +// referenced by any message. +package attachments diff --git a/internal/attachments/mime.go b/internal/attachments/mime.go new file mode 100644 index 0000000..039d2ce --- /dev/null +++ b/internal/attachments/mime.go @@ -0,0 +1,120 @@ +package attachments + +import ( + "net/http" + "path/filepath" + "strings" +) + +// extensionMIMETypes maps file extensions to MIME types for common types +// that net/http.DetectContentType may not identify from magic bytes alone. +var extensionMIMETypes = map[string]string{ + ".css": "text/css", + ".csv": "text/csv", + ".doc": "application/msword", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".gif": "image/gif", + ".html": "text/html", + ".jpeg": "image/jpeg", + ".jpg": "image/jpeg", + ".js": "application/javascript", + ".json": "application/json", + ".md": "text/markdown", + ".mp3": "audio/mpeg", + ".mp4": "video/mp4", + ".pdf": "application/pdf", + ".png": "image/png", + ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", + ".svg": "image/svg+xml", + ".tar": "application/x-tar", + ".txt": "text/plain", + ".wav": "audio/wav", + ".webp": "image/webp", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ".xml": "application/xml", + ".yaml": "text/yaml", + ".yml": "text/yaml", + ".zip": "application/zip", +} + +// imageTypes is the set of MIME types considered "image" for inline preview. +var imageTypes = map[string]bool{ + "image/jpeg": true, + "image/png": true, + "image/gif": true, + "image/webp": true, + "image/svg+xml": true, +} + +// DetectMIMEType detects the MIME type from file content (magic bytes) with +// fallback to extension-based detection. Returns "application/octet-stream" +// if detection fails entirely. +// +// When magic-byte detection returns a generic type like "text/plain" or +// "application/octet-stream", the extension is checked for a more specific +// type (e.g., .csv -> text/csv, .json -> application/json). +func DetectMIMEType(content []byte, filename string) string { + var detected string + if len(content) > 0 { + detected = http.DetectContentType(content) + } + + // If detection returned a specific, non-generic type, use it. + if detected != "" && !isGenericMIME(detected) { + return detected + } + + // Extension-based lookup for more specificity. + if filename != "" { + ext := strings.ToLower(filepath.Ext(filename)) + if mime, ok := extensionMIMETypes[ext]; ok { + return mime + } + } + + // Return whatever detection found, or fall back to octet-stream. + if detected != "" { + return detected + } + return "application/octet-stream" +} + +// isGenericMIME returns true if the MIME type is a generic catch-all that +// should be overridden by extension-based detection when available. +func isGenericMIME(mimeType string) bool { + // Normalize: strip parameters like "; charset=utf-8". + base := mimeType + if idx := strings.Index(mimeType, ";"); idx >= 0 { + base = strings.TrimSpace(mimeType[:idx]) + } + return base == "application/octet-stream" || base == "text/plain" +} + +// DefaultFilename returns a default filename based on MIME type when the +// caller provides none. +func DefaultFilename(mimeType string) string { + switch { + case strings.HasPrefix(mimeType, "image/png"): + return "untitled.png" + case strings.HasPrefix(mimeType, "image/jpeg"): + return "untitled.jpg" + case strings.HasPrefix(mimeType, "image/gif"): + return "untitled.gif" + case strings.HasPrefix(mimeType, "image/webp"): + return "untitled.webp" + case strings.HasPrefix(mimeType, "image/svg"): + return "untitled.svg" + case mimeType == "application/pdf": + return "untitled.pdf" + case strings.HasPrefix(mimeType, "text/"): + return "untitled.txt" + default: + return "untitled.bin" + } +} + +// IsImageType returns true if the MIME type is a supported image type for +// inline preview in the Web UI. +func IsImageType(mimeType string) bool { + return imageTypes[mimeType] +} diff --git a/internal/attachments/mime_test.go b/internal/attachments/mime_test.go new file mode 100644 index 0000000..6df64ef --- /dev/null +++ b/internal/attachments/mime_test.go @@ -0,0 +1,162 @@ +package attachments + +import "testing" + +func TestDetectMIMEType(t *testing.T) { + // PNG magic bytes. + pngHeader := []byte{0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a} + // JPEG magic bytes. + jpegHeader := []byte{0xff, 0xd8, 0xff, 0xe0} + // GIF magic bytes. + gifHeader := []byte("GIF89a") + // PDF magic bytes. + pdfHeader := []byte("%PDF-1.4") + + tests := []struct { + name string + content []byte + filename string + want string + }{ + { + name: "PNG from magic bytes", + content: pngHeader, + filename: "", + want: "image/png", + }, + { + name: "JPEG from magic bytes", + content: jpegHeader, + filename: "", + want: "image/jpeg", + }, + { + name: "GIF from magic bytes", + content: gifHeader, + filename: "", + want: "image/gif", + }, + { + name: "PDF from magic bytes", + content: pdfHeader, + filename: "report.pdf", + want: "application/pdf", + }, + { + name: "CSV from extension", + content: []byte("a,b,c\n1,2,3"), + filename: "data.csv", + want: "text/csv", + }, + { + name: "JSON from extension", + content: []byte(`{"key": "value"}`), + filename: "config.json", + want: "application/json", + }, + { + name: "Markdown from extension", + content: []byte("# Hello"), + filename: "readme.md", + want: "text/markdown", + }, + { + name: "ZIP from extension", + content: []byte("not real zip"), + filename: "archive.zip", + want: "application/zip", + }, + { + name: "unknown falls back to octet-stream", + content: []byte{0x00, 0x01, 0x02}, + filename: "mystery.xyz", + want: "application/octet-stream", + }, + { + name: "empty content with extension", + content: nil, + filename: "test.png", + want: "image/png", + }, + { + name: "empty everything", + content: nil, + filename: "", + want: "application/octet-stream", + }, + { + name: "HTML from magic bytes", + content: []byte("hello"), + filename: "", + want: "text/html; charset=utf-8", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := DetectMIMEType(tt.content, tt.filename) + if got != tt.want { + t.Errorf("DetectMIMEType(%v, %q) = %q, want %q", tt.content[:min(len(tt.content), 8)], tt.filename, got, tt.want) + } + }) + } +} + +func TestDefaultFilename(t *testing.T) { + tests := []struct { + mimeType string + want string + }{ + {"image/png", "untitled.png"}, + {"image/jpeg", "untitled.jpg"}, + {"image/gif", "untitled.gif"}, + {"image/webp", "untitled.webp"}, + {"image/svg+xml", "untitled.svg"}, + {"application/pdf", "untitled.pdf"}, + {"text/plain", "untitled.txt"}, + {"text/csv", "untitled.txt"}, + {"application/octet-stream", "untitled.bin"}, + {"application/zip", "untitled.bin"}, + } + + for _, tt := range tests { + t.Run(tt.mimeType, func(t *testing.T) { + got := DefaultFilename(tt.mimeType) + if got != tt.want { + t.Errorf("DefaultFilename(%q) = %q, want %q", tt.mimeType, got, tt.want) + } + }) + } +} + +func TestIsImageType(t *testing.T) { + tests := []struct { + mimeType string + want bool + }{ + {"image/jpeg", true}, + {"image/png", true}, + {"image/gif", true}, + {"image/webp", true}, + {"image/svg+xml", true}, + {"application/pdf", false}, + {"text/plain", false}, + {"image/tiff", false}, + } + + for _, tt := range tests { + t.Run(tt.mimeType, func(t *testing.T) { + got := IsImageType(tt.mimeType) + if got != tt.want { + t.Errorf("IsImageType(%q) = %v, want %v", tt.mimeType, got, tt.want) + } + }) + } +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} diff --git a/internal/attachments/model.go b/internal/attachments/model.go new file mode 100644 index 0000000..93e693e --- /dev/null +++ b/internal/attachments/model.go @@ -0,0 +1,62 @@ +package attachments + +import ( + "errors" + "io" + "time" +) + +// MaxFileSize is the maximum allowed upload size (50 MB). +const MaxFileSize = 50 * 1024 * 1024 // 50 MB + +// Sentinel errors. +var ( + ErrNotFound = errors.New("attachment not found") + ErrFileTooLarge = errors.New("file exceeds maximum size of 50MB") + ErrEmptyFile = errors.New("empty file not allowed") + ErrFileMissing = errors.New("attachment file missing from disk") +) + +// Attachment represents the metadata for a stored file. +type Attachment struct { + ID int64 `json:"id"` + Hash string `json:"hash"` + OriginalFilename string `json:"original_filename"` + Size int64 `json:"size"` + MIMEType string `json:"mime_type"` + MessageID *int64 `json:"message_id,omitempty"` + UploadedBy string `json:"uploaded_by"` + CreatedAt time.Time `json:"created_at"` +} + +// UploadRequest contains the parameters for uploading an attachment. +type UploadRequest struct { + Content io.Reader + Filename string + MIMEType string + MessageID *int64 + UploadedBy string +} + +// UploadResult contains the result of a successful upload. +type UploadResult struct { + Hash string `json:"hash"` + Size int64 `json:"size"` + MIMEType string `json:"mime_type"` + Filename string `json:"original_filename"` +} + +// DownloadResult contains the result of a successful download. +type DownloadResult struct { + Content io.ReadCloser + Hash string + Filename string + MIMEType string + Size int64 +} + +// GCResult contains the result of a garbage collection run. +type GCResult struct { + FilesRemoved int `json:"files_removed"` + BytesReclaimed int64 `json:"bytes_reclaimed"` +} diff --git a/internal/attachments/service.go b/internal/attachments/service.go new file mode 100644 index 0000000..6866f86 --- /dev/null +++ b/internal/attachments/service.go @@ -0,0 +1,224 @@ +package attachments + +import ( + "bytes" + "context" + "fmt" + "io" + "log/slog" + "sync" + "time" +) + +// Service is the main entry point for all attachment operations. +// It composes the metadata Store with the CAS filesystem engine. +type Service struct { + store Store + cas *CAS + logger *slog.Logger + gcMu sync.Mutex // prevents concurrent GC + upload conflicts +} + +// NewService creates a new attachment service. +func NewService(store Store, cas *CAS, logger *slog.Logger) *Service { + return &Service{ + store: store, + cas: cas, + logger: logger.With("component", "attachment-service"), + } +} + +// Upload stores a file in content-addressable storage and records metadata. +// It validates size, detects MIME type, deduplicates on hash, and logs the +// operation. +func (s *Service) Upload(ctx context.Context, req UploadRequest) (*UploadResult, error) { + start := time.Now() + + // Read the content into a limited reader to enforce size limits. + // We read up to MaxFileSize + 1 to detect overflow. + content, err := io.ReadAll(io.LimitReader(req.Content, MaxFileSize+1)) + if err != nil { + return nil, fmt.Errorf("read content: %w", err) + } + + if len(content) == 0 { + return nil, ErrEmptyFile + } + + if int64(len(content)) > MaxFileSize { + return nil, ErrFileTooLarge + } + + // Detect MIME type if not provided. + mimeType := req.MIMEType + if mimeType == "" { + // Use first 512 bytes for detection. + sniffBuf := content + if len(sniffBuf) > 512 { + sniffBuf = sniffBuf[:512] + } + mimeType = DetectMIMEType(sniffBuf, req.Filename) + } + + // Assign default filename if missing. + filename := req.Filename + if filename == "" { + filename = DefaultFilename(mimeType) + } + + // Write to CAS. + hash, size, err := s.cas.Write(bytes.NewReader(content)) + if err != nil { + return nil, fmt.Errorf("write to CAS: %w", err) + } + + // Store metadata. + att := &Attachment{ + Hash: hash, + OriginalFilename: filename, + Size: size, + MIMEType: mimeType, + MessageID: req.MessageID, + UploadedBy: req.UploadedBy, + } + + if err := s.store.InsertMetadata(ctx, att); err != nil { + return nil, fmt.Errorf("insert metadata: %w", err) + } + + s.logger.Info("attachment uploaded", + "hash", hash, + "filename", filename, + "size", size, + "mime_type", mimeType, + "uploaded_by", req.UploadedBy, + "duration_ms", time.Since(start).Milliseconds(), + ) + + return &UploadResult{ + Hash: hash, + Size: size, + MIMEType: mimeType, + Filename: filename, + }, nil +} + +// Download retrieves a file and its metadata by hash. +// Returns ErrNotFound if no metadata exists, or ErrFileMissing if the +// metadata exists but the file is not on disk. +func (s *Service) Download(ctx context.Context, hash string) (*DownloadResult, error) { + start := time.Now() + + // Look up metadata. + atts, err := s.store.GetByHash(ctx, hash) + if err != nil { + return nil, fmt.Errorf("get metadata: %w", err) + } + if len(atts) == 0 { + return nil, ErrNotFound + } + + // Use the first metadata row for filename/mime info. + meta := atts[0] + + // Open the file from CAS. + reader, err := s.cas.Read(hash) + if err != nil { + if err == ErrNotFound { + s.logger.Warn("attachment file missing from disk", + "hash", hash, + "filename", meta.OriginalFilename, + ) + return nil, ErrFileMissing + } + return nil, fmt.Errorf("read from CAS: %w", err) + } + + s.logger.Info("attachment downloaded", + "hash", hash, + "filename", meta.OriginalFilename, + "size", meta.Size, + "duration_ms", time.Since(start).Milliseconds(), + ) + + return &DownloadResult{ + Content: reader, + Hash: hash, + Filename: meta.OriginalFilename, + MIMEType: meta.MIMEType, + Size: meta.Size, + }, nil +} + +// AttachToMessage links an existing attachment to a message by updating +// the message_id on all metadata rows with the given hash that are currently +// unlinked. +func (s *Service) AttachToMessage(ctx context.Context, hash string, messageID int64) error { + atts, err := s.store.GetByHash(ctx, hash) + if err != nil { + return fmt.Errorf("get metadata: %w", err) + } + if len(atts) == 0 { + return ErrNotFound + } + + // We update by inserting a new metadata row linked to the message, + // copying from the first existing row. + meta := atts[0] + linked := &Attachment{ + Hash: meta.Hash, + OriginalFilename: meta.OriginalFilename, + Size: meta.Size, + MIMEType: meta.MIMEType, + MessageID: &messageID, + UploadedBy: meta.UploadedBy, + } + return s.store.InsertMetadata(ctx, linked) +} + +// GetByMessageID returns all attachments for a given message. +func (s *Service) GetByMessageID(ctx context.Context, messageID int64) ([]*Attachment, error) { + return s.store.GetByMessageID(ctx, messageID) +} + +// GarbageCollect finds and removes orphaned attachments — files that are +// no longer referenced by any message. Returns a summary of what was removed. +func (s *Service) GarbageCollect(ctx context.Context) (*GCResult, error) { + s.gcMu.Lock() + defer s.gcMu.Unlock() + + start := time.Now() + + orphans, err := s.store.FindOrphanHashes(ctx) + if err != nil { + return nil, fmt.Errorf("find orphans: %w", err) + } + + result := &GCResult{} + + for _, hash := range orphans { + // Delete from CAS. + size, err := s.cas.Delete(hash) + if err != nil { + s.logger.Error("GC: failed to delete file", "hash", hash, "error", err) + continue + } + + // Delete metadata. + if err := s.store.DeleteByHash(ctx, hash); err != nil { + s.logger.Error("GC: failed to delete metadata", "hash", hash, "error", err) + continue + } + + result.FilesRemoved++ + result.BytesReclaimed += size + } + + s.logger.Info("garbage collection complete", + "files_removed", result.FilesRemoved, + "bytes_reclaimed", result.BytesReclaimed, + "duration_ms", time.Since(start).Milliseconds(), + ) + + return result, nil +} diff --git a/internal/attachments/service_test.go b/internal/attachments/service_test.go new file mode 100644 index 0000000..5092dcf --- /dev/null +++ b/internal/attachments/service_test.go @@ -0,0 +1,265 @@ +package attachments + +import ( + "bytes" + "context" + "database/sql" + "io" + "log/slog" + "path/filepath" + "strings" + "testing" +) + +func newTestService(t *testing.T) (*Service, *sql.DB) { + t.Helper() + db := newTestDB(t) + dir := t.TempDir() + + cas, err := NewCAS(filepath.Join(dir, "attachments"), slog.Default()) + if err != nil { + t.Fatalf("NewCAS: %v", err) + } + + store := NewSQLiteStore(db, slog.Default()) + svc := NewService(store, cas, slog.Default()) + return svc, db +} + +func TestService_Upload(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + content := []byte("service upload test content") + wantHash := sha256Hex(content) + + result, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(content), + Filename: "test.txt", + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload: %v", err) + } + + if result.Hash != wantHash { + t.Errorf("hash = %s, want %s", result.Hash, wantHash) + } + if result.Size != int64(len(content)) { + t.Errorf("size = %d, want %d", result.Size, len(content)) + } + if result.Filename != "test.txt" { + t.Errorf("filename = %s, want test.txt", result.Filename) + } +} + +func TestService_Upload_MIMEDetection(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + // PNG magic bytes. + pngContent := []byte{0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00} + + result, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(pngContent), + Filename: "image.png", + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload: %v", err) + } + + if result.MIMEType != "image/png" { + t.Errorf("mime_type = %s, want image/png", result.MIMEType) + } +} + +func TestService_Upload_DefaultFilename(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + content := []byte("no filename content here") + + result, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(content), + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload: %v", err) + } + + if result.Filename == "" { + t.Error("expected a default filename, got empty") + } +} + +func TestService_Upload_EmptyFile(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + _, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader([]byte{}), + Filename: "empty.txt", + UploadedBy: "agent-a", + }) + if err != ErrEmptyFile { + t.Errorf("expected ErrEmptyFile, got %v", err) + } +} + +func TestService_Upload_SizeLimit(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + // Create a reader that yields MaxFileSize+1 bytes. + oversized := make([]byte, MaxFileSize+1) + for i := range oversized { + oversized[i] = 'x' + } + + _, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(oversized), + Filename: "toobig.bin", + UploadedBy: "agent-a", + }) + if err != ErrFileTooLarge { + t.Errorf("expected ErrFileTooLarge, got %v", err) + } +} + +func TestService_Download(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + content := []byte("download test content") + + uploadResult, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(content), + Filename: "download.txt", + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload: %v", err) + } + + t.Run("successful download", func(t *testing.T) { + result, err := svc.Download(ctx, uploadResult.Hash) + if err != nil { + t.Fatalf("Download: %v", err) + } + defer result.Content.Close() + + got, err := io.ReadAll(result.Content) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !bytes.Equal(got, content) { + t.Error("downloaded content does not match uploaded content") + } + if result.Filename != "download.txt" { + t.Errorf("filename = %s, want download.txt", result.Filename) + } + }) + + t.Run("not found", func(t *testing.T) { + _, err := svc.Download(ctx, "deadbeef"+strings.Repeat("0", 56)) + if err != ErrNotFound { + t.Errorf("expected ErrNotFound, got %v", err) + } + }) +} + +func TestService_Dedup(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + content := []byte("dedup test content") + wantHash := sha256Hex(content) + + r1, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(content), + Filename: "version1.txt", + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload 1: %v", err) + } + + r2, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(content), + Filename: "version2.txt", + UploadedBy: "agent-b", + }) + if err != nil { + t.Fatalf("Upload 2: %v", err) + } + + if r1.Hash != r2.Hash { + t.Errorf("hashes differ: %s vs %s", r1.Hash, r2.Hash) + } + if r1.Hash != wantHash { + t.Errorf("hash = %s, want %s", r1.Hash, wantHash) + } +} + +func TestService_GarbageCollect(t *testing.T) { + svc, db := newTestService(t) + ctx := context.Background() + + // Upload an orphan (no message_id). + orphanContent := []byte("orphan gc content") + _, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(orphanContent), + Filename: "orphan.txt", + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload orphan: %v", err) + } + + // Upload a linked attachment. + msgID := seedMessage(t, db) + linkedContent := []byte("linked gc content") + linkedResult, err := svc.Upload(ctx, UploadRequest{ + Content: bytes.NewReader(linkedContent), + Filename: "linked.txt", + MessageID: &msgID, + UploadedBy: "agent-a", + }) + if err != nil { + t.Fatalf("Upload linked: %v", err) + } + + // Run GC. + gc, err := svc.GarbageCollect(ctx) + if err != nil { + t.Fatalf("GarbageCollect: %v", err) + } + + if gc.FilesRemoved != 1 { + t.Errorf("files_removed = %d, want 1", gc.FilesRemoved) + } + + // Verify linked file still downloadable. + result, err := svc.Download(ctx, linkedResult.Hash) + if err != nil { + t.Fatalf("Download linked after GC: %v", err) + } + result.Content.Close() +} + +func TestService_GarbageCollect_Empty(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + + gc, err := svc.GarbageCollect(ctx) + if err != nil { + t.Fatalf("GarbageCollect: %v", err) + } + if gc.FilesRemoved != 0 { + t.Errorf("files_removed = %d, want 0", gc.FilesRemoved) + } + if gc.BytesReclaimed != 0 { + t.Errorf("bytes_reclaimed = %d, want 0", gc.BytesReclaimed) + } +} diff --git a/internal/attachments/sqlite_store.go b/internal/attachments/sqlite_store.go new file mode 100644 index 0000000..ef118a4 --- /dev/null +++ b/internal/attachments/sqlite_store.go @@ -0,0 +1,167 @@ +package attachments + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "time" +) + +// SQLiteStore implements Store backed by modernc.org/sqlite. +type SQLiteStore struct { + db *sql.DB + logger *slog.Logger +} + +// NewSQLiteStore creates a new SQLite-backed attachment metadata store. +func NewSQLiteStore(db *sql.DB, logger *slog.Logger) *SQLiteStore { + return &SQLiteStore{ + db: db, + logger: logger.With("component", "attachment-store"), + } +} + +// InsertMetadata inserts a new attachment metadata row. +func (s *SQLiteStore) InsertMetadata(ctx context.Context, a *Attachment) error { + const query = `INSERT INTO attachments (hash, original_filename, size, mime_type, message_id, uploaded_by, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)` + + now := time.Now().UTC() + result, err := s.db.ExecContext(ctx, query, + a.Hash, + a.OriginalFilename, + a.Size, + a.MIMEType, + a.MessageID, + a.UploadedBy, + now, + ) + if err != nil { + return fmt.Errorf("insert attachment metadata: %w", err) + } + + id, err := result.LastInsertId() + if err != nil { + return fmt.Errorf("get last insert id: %w", err) + } + + a.ID = id + a.CreatedAt = now + + s.logger.Debug("attachment metadata inserted", + "id", id, + "hash", a.Hash, + "filename", a.OriginalFilename, + ) + return nil +} + +// GetByHash returns all attachment metadata rows matching the given hash. +func (s *SQLiteStore) GetByHash(ctx context.Context, hash string) ([]*Attachment, error) { + const query = `SELECT id, hash, original_filename, size, mime_type, message_id, uploaded_by, created_at + FROM attachments WHERE hash = ? ORDER BY created_at DESC` + + rows, err := s.db.QueryContext(ctx, query, hash) + if err != nil { + return nil, fmt.Errorf("query attachments by hash: %w", err) + } + defer rows.Close() + + return scanAttachments(rows) +} + +// GetByMessageID returns all attachment metadata rows for a given message. +func (s *SQLiteStore) GetByMessageID(ctx context.Context, messageID int64) ([]*Attachment, error) { + const query = `SELECT id, hash, original_filename, size, mime_type, message_id, uploaded_by, created_at + FROM attachments WHERE message_id = ? ORDER BY created_at ASC` + + rows, err := s.db.QueryContext(ctx, query, messageID) + if err != nil { + return nil, fmt.Errorf("query attachments by message_id: %w", err) + } + defer rows.Close() + + return scanAttachments(rows) +} + +// DeleteByHash removes all metadata rows matching the given hash. +func (s *SQLiteStore) DeleteByHash(ctx context.Context, hash string) error { + const query = `DELETE FROM attachments WHERE hash = ?` + + result, err := s.db.ExecContext(ctx, query, hash) + if err != nil { + return fmt.Errorf("delete attachment metadata: %w", err) + } + + n, _ := result.RowsAffected() + s.logger.Debug("attachment metadata deleted", "hash", hash, "rows", n) + return nil +} + +// FindOrphanHashes returns hashes that have no valid message reference. +// A hash is orphaned if all its metadata rows have message_id IS NULL or +// the referenced message no longer exists. +func (s *SQLiteStore) FindOrphanHashes(ctx context.Context) ([]string, error) { + const query = `SELECT DISTINCT a.hash FROM attachments a + WHERE a.message_id IS NULL + OR a.message_id NOT IN (SELECT id FROM messages) + GROUP BY a.hash + HAVING COUNT(CASE WHEN a.message_id IN (SELECT id FROM messages) THEN 1 END) = 0` + + rows, err := s.db.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("find orphan hashes: %w", err) + } + defer rows.Close() + + var hashes []string + for rows.Next() { + var h string + if err := rows.Scan(&h); err != nil { + return nil, fmt.Errorf("scan orphan hash: %w", err) + } + hashes = append(hashes, h) + } + return hashes, rows.Err() +} + +// CountReferences returns the number of metadata rows referencing the hash. +func (s *SQLiteStore) CountReferences(ctx context.Context, hash string) (int64, error) { + const query = `SELECT COUNT(*) FROM attachments WHERE hash = ?` + + var count int64 + if err := s.db.QueryRowContext(ctx, query, hash).Scan(&count); err != nil { + return 0, fmt.Errorf("count references: %w", err) + } + return count, nil +} + +// scanAttachments scans rows into a slice of Attachment pointers. +func scanAttachments(rows *sql.Rows) ([]*Attachment, error) { + var attachments []*Attachment + for rows.Next() { + a := &Attachment{} + var createdAtStr string + if err := rows.Scan( + &a.ID, + &a.Hash, + &a.OriginalFilename, + &a.Size, + &a.MIMEType, + &a.MessageID, + &a.UploadedBy, + &createdAtStr, + ); err != nil { + return nil, fmt.Errorf("scan attachment: %w", err) + } + // Parse the created_at timestamp. + if t, err := time.Parse("2006-01-02 15:04:05", createdAtStr); err == nil { + a.CreatedAt = t + } else if t, err := time.Parse(time.RFC3339, createdAtStr); err == nil { + a.CreatedAt = t + } + attachments = append(attachments, a) + } + return attachments, rows.Err() +} diff --git a/internal/attachments/sqlite_store_test.go b/internal/attachments/sqlite_store_test.go new file mode 100644 index 0000000..3620040 --- /dev/null +++ b/internal/attachments/sqlite_store_test.go @@ -0,0 +1,226 @@ +package attachments + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "testing" + + _ "modernc.org/sqlite" + + "github.com/smart-mcp-proxy/synapbus/internal/storage" +) + +func newTestDB(t *testing.T) *sql.DB { + t.Helper() + dsn := fmt.Sprintf("file:%s?mode=memory&cache=shared", t.Name()) + db, err := sql.Open("sqlite", dsn) + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { db.Close() }) + + if _, err := db.Exec("PRAGMA foreign_keys=ON"); err != nil { + t.Fatalf("enable foreign keys: %v", err) + } + + ctx := context.Background() + if err := storage.RunMigrations(ctx, db); err != nil { + t.Fatalf("run migrations: %v", err) + } + + // Seed test user and agent. + db.Exec(`INSERT OR IGNORE INTO users (id, username, password_hash, display_name) VALUES (1, 'testowner', 'hash', 'Test Owner')`) + + return db +} + +func newTestStore(t *testing.T) (*SQLiteStore, *sql.DB) { + t.Helper() + db := newTestDB(t) + store := NewSQLiteStore(db, slog.Default()) + return store, db +} + +// seedMessage inserts a test message so we can reference it via foreign key. +func seedMessage(t *testing.T, db *sql.DB) int64 { + t.Helper() + // Ensure user exists. + db.Exec(`INSERT OR IGNORE INTO users (id, username, password_hash, display_name) VALUES (1, 'testowner', 'hash', 'Test Owner')`) + // Ensure agents exist. + db.Exec(`INSERT OR IGNORE INTO agents (id, name, display_name, owner_id, api_key_hash) VALUES (1, 'agent-a', 'Agent A', 1, 'hash')`) + db.Exec(`INSERT OR IGNORE INTO agents (id, name, display_name, owner_id, api_key_hash) VALUES (2, 'agent-b', 'Agent B', 1, 'hash')`) + // Ensure conversation exists. + db.Exec(`INSERT OR IGNORE INTO conversations (id, subject, created_by) VALUES (1, 'test', 'agent-a')`) + // Insert message. + result, err := db.Exec(`INSERT INTO messages (conversation_id, from_agent, to_agent, body) VALUES (1, 'agent-a', 'agent-b', 'hello')`) + if err != nil { + t.Fatalf("seed message: %v", err) + } + id, _ := result.LastInsertId() + return id +} + +func TestSQLiteStore_InsertMetadata(t *testing.T) { + store, _ := newTestStore(t) + ctx := context.Background() + + a := &Attachment{ + Hash: "abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890", + OriginalFilename: "test.png", + Size: 1024, + MIMEType: "image/png", + UploadedBy: "agent-a", + } + + if err := store.InsertMetadata(ctx, a); err != nil { + t.Fatalf("InsertMetadata: %v", err) + } + + if a.ID == 0 { + t.Error("ID not populated after insert") + } + if a.CreatedAt.IsZero() { + t.Error("CreatedAt not populated after insert") + } +} + +func TestSQLiteStore_GetByHash(t *testing.T) { + store, _ := newTestStore(t) + ctx := context.Background() + + hash := "abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890" + + // Insert two rows with the same hash. + a1 := &Attachment{Hash: hash, OriginalFilename: "file1.png", Size: 100, MIMEType: "image/png", UploadedBy: "agent-a"} + a2 := &Attachment{Hash: hash, OriginalFilename: "file2.png", Size: 100, MIMEType: "image/png", UploadedBy: "agent-b"} + + if err := store.InsertMetadata(ctx, a1); err != nil { + t.Fatalf("InsertMetadata a1: %v", err) + } + if err := store.InsertMetadata(ctx, a2); err != nil { + t.Fatalf("InsertMetadata a2: %v", err) + } + + results, err := store.GetByHash(ctx, hash) + if err != nil { + t.Fatalf("GetByHash: %v", err) + } + if len(results) != 2 { + t.Fatalf("expected 2 results, got %d", len(results)) + } +} + +func TestSQLiteStore_GetByMessageID(t *testing.T) { + store, db := newTestStore(t) + ctx := context.Background() + + msgID := seedMessage(t, db) + + a := &Attachment{ + Hash: "1111111111111111111111111111111111111111111111111111111111111111", + OriginalFilename: "report.pdf", + Size: 2048, + MIMEType: "application/pdf", + MessageID: &msgID, + UploadedBy: "agent-a", + } + if err := store.InsertMetadata(ctx, a); err != nil { + t.Fatalf("InsertMetadata: %v", err) + } + + results, err := store.GetByMessageID(ctx, msgID) + if err != nil { + t.Fatalf("GetByMessageID: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected 1 result, got %d", len(results)) + } + if results[0].OriginalFilename != "report.pdf" { + t.Errorf("filename = %s, want report.pdf", results[0].OriginalFilename) + } +} + +func TestSQLiteStore_DeleteByHash(t *testing.T) { + store, _ := newTestStore(t) + ctx := context.Background() + + hash := "2222222222222222222222222222222222222222222222222222222222222222" + a := &Attachment{Hash: hash, OriginalFilename: "del.txt", Size: 10, MIMEType: "text/plain", UploadedBy: "agent-a"} + if err := store.InsertMetadata(ctx, a); err != nil { + t.Fatalf("InsertMetadata: %v", err) + } + + if err := store.DeleteByHash(ctx, hash); err != nil { + t.Fatalf("DeleteByHash: %v", err) + } + + results, err := store.GetByHash(ctx, hash) + if err != nil { + t.Fatalf("GetByHash after delete: %v", err) + } + if len(results) != 0 { + t.Errorf("expected 0 results after delete, got %d", len(results)) + } +} + +func TestSQLiteStore_CountReferences(t *testing.T) { + store, _ := newTestStore(t) + ctx := context.Background() + + hash := "3333333333333333333333333333333333333333333333333333333333333333" + + count, err := store.CountReferences(ctx, hash) + if err != nil { + t.Fatalf("CountReferences: %v", err) + } + if count != 0 { + t.Errorf("expected 0 references, got %d", count) + } + + a := &Attachment{Hash: hash, OriginalFilename: "ref.txt", Size: 5, MIMEType: "text/plain", UploadedBy: "agent-a"} + if err := store.InsertMetadata(ctx, a); err != nil { + t.Fatalf("InsertMetadata: %v", err) + } + + count, err = store.CountReferences(ctx, hash) + if err != nil { + t.Fatalf("CountReferences: %v", err) + } + if count != 1 { + t.Errorf("expected 1 reference, got %d", count) + } +} + +func TestSQLiteStore_FindOrphanHashes(t *testing.T) { + store, db := newTestStore(t) + ctx := context.Background() + + // Create an orphan attachment (no message_id). + orphanHash := "4444444444444444444444444444444444444444444444444444444444444444" + a := &Attachment{Hash: orphanHash, OriginalFilename: "orphan.txt", Size: 5, MIMEType: "text/plain", UploadedBy: "agent-a"} + if err := store.InsertMetadata(ctx, a); err != nil { + t.Fatalf("InsertMetadata orphan: %v", err) + } + + // Create a linked attachment. + msgID := seedMessage(t, db) + linkedHash := "5555555555555555555555555555555555555555555555555555555555555555" + aLinked := &Attachment{Hash: linkedHash, OriginalFilename: "linked.txt", Size: 5, MIMEType: "text/plain", MessageID: &msgID, UploadedBy: "agent-a"} + if err := store.InsertMetadata(ctx, aLinked); err != nil { + t.Fatalf("InsertMetadata linked: %v", err) + } + + orphans, err := store.FindOrphanHashes(ctx) + if err != nil { + t.Fatalf("FindOrphanHashes: %v", err) + } + + if len(orphans) != 1 { + t.Fatalf("expected 1 orphan, got %d", len(orphans)) + } + if orphans[0] != orphanHash { + t.Errorf("orphan hash = %s, want %s", orphans[0], orphanHash) + } +} diff --git a/internal/attachments/store.go b/internal/attachments/store.go new file mode 100644 index 0000000..4459dcb --- /dev/null +++ b/internal/attachments/store.go @@ -0,0 +1,25 @@ +package attachments + +import "context" + +// Store is the metadata persistence interface for attachments. +type Store interface { + // InsertMetadata inserts a new attachment metadata row. + InsertMetadata(ctx context.Context, a *Attachment) error + + // GetByHash returns all attachment metadata rows matching the given hash. + GetByHash(ctx context.Context, hash string) ([]*Attachment, error) + + // GetByMessageID returns all attachment metadata rows for a given message. + GetByMessageID(ctx context.Context, messageID int64) ([]*Attachment, error) + + // DeleteByHash removes all metadata rows matching the given hash. + DeleteByHash(ctx context.Context, hash string) error + + // FindOrphanHashes returns hashes that have no message_id reference + // (message_id IS NULL or the referenced message no longer exists). + FindOrphanHashes(ctx context.Context) ([]string, error) + + // CountReferences returns the number of metadata rows referencing the hash. + CountReferences(ctx context.Context, hash string) (int64, error) +} diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 231a3c7..4f6d101 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -8,6 +8,7 @@ import ( "github.com/mark3labs/mcp-go/server" "github.com/smart-mcp-proxy/synapbus/internal/agents" + "github.com/smart-mcp-proxy/synapbus/internal/attachments" "github.com/smart-mcp-proxy/synapbus/internal/channels" "github.com/smart-mcp-proxy/synapbus/internal/messaging" ) @@ -26,6 +27,7 @@ func NewMCPServer( msgService *messaging.MessagingService, agentService *agents.AgentService, channelService *channels.Service, + attachmentService *attachments.Service, ) *MCPServer { logger := slog.Default().With("component", "mcp-server") @@ -46,6 +48,12 @@ func NewMCPServer( channelRegistrar.RegisterAll(mcpSrv) } + // Register attachment tools + if attachmentService != nil { + attachmentRegistrar := NewAttachmentToolRegistrar(attachmentService) + attachmentRegistrar.RegisterAll(mcpSrv) + } + // Create SSE transport with context func for auth propagation sseServer := server.NewSSEServer(mcpSrv, server.WithSSEContextFunc(func(ctx context.Context, r *http.Request) context.Context { diff --git a/internal/mcp/tools_attachments.go b/internal/mcp/tools_attachments.go new file mode 100644 index 0000000..c5dd923 --- /dev/null +++ b/internal/mcp/tools_attachments.go @@ -0,0 +1,161 @@ +package mcp + +import ( + "bytes" + "context" + "encoding/base64" + "fmt" + "io" + + "github.com/mark3labs/mcp-go/mcp" + "github.com/mark3labs/mcp-go/server" + + "github.com/smart-mcp-proxy/synapbus/internal/attachments" +) + +// AttachmentToolRegistrar registers attachment MCP tools on the server. +type AttachmentToolRegistrar struct { + attachmentService *attachments.Service +} + +// NewAttachmentToolRegistrar creates a new attachment tool registrar. +func NewAttachmentToolRegistrar(attachmentService *attachments.Service) *AttachmentToolRegistrar { + return &AttachmentToolRegistrar{ + attachmentService: attachmentService, + } +} + +// RegisterAll registers all attachment tools on the MCP server. +func (atr *AttachmentToolRegistrar) RegisterAll(s *server.MCPServer) { + s.AddTool(atr.uploadAttachmentTool(), atr.handleUploadAttachment) + s.AddTool(atr.downloadAttachmentTool(), atr.handleDownloadAttachment) + s.AddTool(atr.gcAttachmentsTool(), atr.handleGCAttachments) +} + +// --- Tool Definitions --- + +func (atr *AttachmentToolRegistrar) uploadAttachmentTool() mcp.Tool { + return mcp.NewTool("upload_attachment", + mcp.WithDescription("Upload a file attachment. Content must be base64-encoded. Returns the SHA-256 hash for later retrieval. Max file size: 50MB."), + mcp.WithString("content", mcp.Description("Base64-encoded file content"), mcp.Required()), + mcp.WithString("filename", mcp.Description("Original filename (optional, used for MIME detection and display)")), + mcp.WithString("mime_type", mcp.Description("MIME type override (optional, auto-detected from content if not provided)")), + mcp.WithNumber("message_id", mcp.Description("Message ID to attach the file to (optional, can be linked later)")), + ) +} + +func (atr *AttachmentToolRegistrar) downloadAttachmentTool() mcp.Tool { + return mcp.NewTool("download_attachment", + mcp.WithDescription("Download an attachment by its SHA-256 hash. Returns base64-encoded content along with filename and MIME type metadata."), + mcp.WithString("hash", mcp.Description("SHA-256 hash of the attachment"), mcp.Required()), + ) +} + +func (atr *AttachmentToolRegistrar) gcAttachmentsTool() mcp.Tool { + return mcp.NewTool("gc_attachments", + mcp.WithDescription("Run garbage collection to remove orphaned attachments not referenced by any message. Returns a summary of files removed and bytes reclaimed."), + ) +} + +// --- Tool Handlers --- + +func (atr *AttachmentToolRegistrar) handleUploadAttachment(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + agentName, ok := extractAgentName(ctx) + if !ok { + return mcp.NewToolResultError("authentication required"), nil + } + + contentB64 := req.GetString("content", "") + if contentB64 == "" { + return mcp.NewToolResultError("'content' parameter is required"), nil + } + + // Check base64 size before decoding to avoid buffering oversized content. + // Base64 expands data by ~4/3, so decoded size is roughly 3/4 of encoded. + if int64(len(contentB64))*3/4 > attachments.MaxFileSize { + return mcp.NewToolResultError("file exceeds maximum size of 50MB"), nil + } + + decoded, err := base64.StdEncoding.DecodeString(contentB64) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("invalid base64 content: %s", err)), nil + } + + if int64(len(decoded)) > attachments.MaxFileSize { + return mcp.NewToolResultError("file exceeds maximum size of 50MB"), nil + } + + uploadReq := attachments.UploadRequest{ + Content: bytes.NewReader(decoded), + Filename: req.GetString("filename", ""), + MIMEType: req.GetString("mime_type", ""), + UploadedBy: agentName, + } + + // Optional message_id. + if mid := req.GetInt("message_id", 0); mid > 0 { + v := int64(mid) + uploadReq.MessageID = &v + } + + result, err := atr.attachmentService.Upload(ctx, uploadReq) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("upload_attachment failed: %s", err)), nil + } + + return resultJSON(map[string]any{ + "hash": result.Hash, + "size": result.Size, + "mime_type": result.MIMEType, + "original_filename": result.Filename, + }) +} + +func (atr *AttachmentToolRegistrar) handleDownloadAttachment(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + _, ok := extractAgentName(ctx) + if !ok { + return mcp.NewToolResultError("authentication required"), nil + } + + hash := req.GetString("hash", "") + if hash == "" { + return mcp.NewToolResultError("'hash' parameter is required"), nil + } + + result, err := atr.attachmentService.Download(ctx, hash) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("download_attachment failed: %s", err)), nil + } + defer result.Content.Close() + + // Read content and base64-encode it. + content, err := io.ReadAll(result.Content) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("read attachment content failed: %s", err)), nil + } + + return resultJSON(map[string]any{ + "hash": result.Hash, + "content": base64.StdEncoding.EncodeToString(content), + "original_filename": result.Filename, + "mime_type": result.MIMEType, + "size": result.Size, + }) +} + +func (atr *AttachmentToolRegistrar) handleGCAttachments(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + _, ok := extractAgentName(ctx) + if !ok { + return mcp.NewToolResultError("authentication required"), nil + } + + result, err := atr.attachmentService.GarbageCollect(ctx) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("gc_attachments failed: %s", err)), nil + } + + return resultJSON(map[string]any{ + "files_removed": result.FilesRemoved, + "bytes_reclaimed": result.BytesReclaimed, + }) +}