amadan

eleven / messenger Public

server: put attachment bytes behind a blobStore seam (#550 Phase 0) (#636)

Attachments now flow through a small blobStore interface (staged
create/commit/abort writes, open with size+modtime, best-effort remove)
instead of reaching for the filesystem directly. diskBlobStore wraps
exactly today's storage — <db>-blobs/ next to the SQLite file, unfetchable
.part staging, atomic rename on publish — so behavior, limits, and auth
are unchanged. Metadata (size for quota, thread tag for GC) stays in the
blobs table; the store holds bytes only and never learns filenames or
MIME. The seam is what later lets the flagship point media at an object
store (issue #550 decision 9) while self-hosters keep on-disk storage;
the legacy DB-column fallback for pre-disk blobs is untouched.

Adds an interface-level round-trip test (put -> open -> equal -> delete
-> open fails) against implementation #1; existing attachment tests pass
unchanged (blobPath became test-only sugar).
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 20257d7f30eb8536477e43a67f5f48ad4672907f parent 3c64c06
4 files changed, +265 −51
  • blobstore.go +127 −0
  • blobstore_test.go +91 −0
  • db.go +31 −28
  • main.go +16 −23
diff --git a/blobstore.go b/blobstore.go
new file mode 100644
index 0000000..fb17b57
--- /dev/null
+++ b/blobstore.go
@@ -0,0 +1,127 @@
+package main
+
+import (
+ "errors"
+ "io"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "time"
+)
+
+// blobStore is the seam between the messenger and wherever attachment bytes
+// physically live. It exists so the flagship can later point media at an
+// object store (issue #550) while self-hosters keep today's on-disk layout —
+// the handlers and GC talk only to this interface.
+//
+// The store holds BYTES ONLY. Everything the server knows *about* a blob —
+// its size (quota accounting) and thread tag (thread-scoped GC) — stays in
+// SQLite's blobs table, which travels with the rest of the instance data.
+// Content arrives as client-side ciphertext; the store never learns filenames,
+// MIME types, or plaintext (those ride inside the encrypted message payload,
+// per the E2EE rule). Objects are write-once and immutable: an id is minted
+// exactly once (randToken), never rewritten, only ever deleted — so there is
+// no versioning or CAS here, unlike home's vaultStore.
+type blobStore interface {
+ // create begins a staged write. The staged bytes are invisible to open
+ // until commit publishes them under an id, so a partial or abandoned
+ // upload can never be served. The two-step shape mirrors what the upload
+ // handler actually does: it learns the size only after streaming the
+ // body, and must apply the quota verdict between "fully received" and
+ // "fetchable".
+ create() (blobWriter, error)
+ // open returns a reader over a committed blob's bytes, plus its size and
+ // modification time (http.ServeContent wants seek + modtime for Range and
+ // conditional requests — video playback scrubs). An id the store doesn't
+ // hold returns an error satisfying errors.Is(err, fs.ErrNotExist).
+ open(id string) (r io.ReadSeekCloser, size int64, modTime time.Time, err error)
+ // remove deletes a blob's bytes. Removing an id that isn't there is not
+ // an error — GC call sites unlink best-effort.
+ remove(id string) error
+}
+
+// blobWriter is one staged write. Exactly one of commit or abort ends it.
+type blobWriter interface {
+ io.Writer
+ // commit publishes the staged bytes under id: after commit, open(id)
+ // serves exactly what was written. Publication is atomic — a crash
+ // mid-commit leaves either nothing or the whole blob, never a readable
+ // partial.
+ commit(id string) error
+ // abort discards the staged bytes. Best-effort with no error to report:
+ // every caller is already unwinding a failure.
+ abort()
+}
+
+// diskBlobStore is implementation #1: one file per blob id under a single
+// directory. It wraps exactly the storage that predates the seam — the dir
+// sits next to the DB file (<db>-blobs/) so it's per-instance and a data
+// snapshot captures both by copying the DB and its sibling dir. Large media
+// never bloats the SQLite file itself (keeping WAL/VACUUM/snapshots lean).
+type diskBlobStore struct{ dir string }
+
+func newDiskBlobStore(dir string) (*diskBlobStore, error) {
+ if err := os.MkdirAll(dir, 0o755); err != nil {
+ return nil, err
+ }
+ return &diskBlobStore{dir: dir}, nil
+}
+
+// path is the on-disk location of a blob's bytes. Blob ids are randToken
+// (base64url), so they're always safe filename components — but the HTTP
+// handler still validates the id (validBlobID) before it can reach here.
+func (s *diskBlobStore) path(id string) string { return filepath.Join(s.dir, id) }
+
+func (s *diskBlobStore) create() (blobWriter, error) {
+ // The "." in the temp name makes it an invalid blob id (validBlobID), so
+ // an in-flight staged write can never be fetched via /api/blob. Same dir
+ // as the final file so the commit rename is atomic (same filesystem).
+ f, err := os.CreateTemp(s.dir, "up.*.part")
+ if err != nil {
+ return nil, err
+ }
+ return &diskBlobWriter{f: f, store: s}, nil
+}
+
+func (s *diskBlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, error) {
+ f, err := os.Open(s.path(id))
+ if err != nil {
+ return nil, 0, time.Time{}, err // wraps fs.ErrNotExist for a missing id
+ }
+ fi, err := f.Stat()
+ if err != nil {
+ f.Close()
+ return nil, 0, time.Time{}, err
+ }
+ return f, fi.Size(), fi.ModTime(), nil
+}
+
+func (s *diskBlobStore) remove(id string) error {
+ if err := os.Remove(s.path(id)); err != nil && !errors.Is(err, fs.ErrNotExist) {
+ return err
+ }
+ return nil
+}
+
+type diskBlobWriter struct {
+ f *os.File
+ store *diskBlobStore
+}
+
+func (w *diskBlobWriter) Write(p []byte) (int, error) { return w.f.Write(p) }
+
+func (w *diskBlobWriter) commit(id string) error {
+ // Close error deliberately ignored, matching the pre-seam upload path:
+ // the copy's own error return is the signal callers act on.
+ w.f.Close()
+ if err := os.Rename(w.f.Name(), w.store.path(id)); err != nil {
+ os.Remove(w.f.Name())
+ return err
+ }
+ return nil
+}
+
+func (w *diskBlobWriter) abort() {
+ w.f.Close()
+ os.Remove(w.f.Name())
+}
diff --git a/blobstore_test.go b/blobstore_test.go
new file mode 100644
index 0000000..739fd9c
--- /dev/null
+++ b/blobstore_test.go
@@ -0,0 +1,91 @@
+package main
+
+import (
+ "bytes"
+ "errors"
+ "io"
+ "io/fs"
+ "path/filepath"
+ "testing"
+)
+
+// blobPath points tests at the default disk store's file for a blob id —
+// test-only sugar (asserting bytes really live on disk / were unlinked).
+// Production code goes through db.blobs and must not assume bytes are files.
+func (db *DB) blobPath(id string) string { return db.blobs.(*diskBlobStore).path(id) }
+
+// The blobStore contract, exercised against implementation #1 (disk): a staged
+// write committed under an id round-trips byte-for-byte, and remove makes it
+// unopenable. Any future implementation (an object store, per issue #550) must
+// pass this same sequence.
+func TestDiskBlobStoreRoundTrip(t *testing.T) {
+ store, err := newDiskBlobStore(filepath.Join(t.TempDir(), "blobs"))
+ if err != nil {
+ t.Fatalf("newDiskBlobStore: %v", err)
+ }
+ payload := []byte("pretend-this-is-encrypted-bytes\x00\x01\x02")
+
+ // put: stream in two writes to prove staging accumulates, then publish.
+ w, err := store.create()
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+ if _, err := w.Write(payload[:10]); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ if _, err := w.Write(payload[10:]); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ if err := w.commit("blobid_123-abc"); err != nil {
+ t.Fatalf("commit: %v", err)
+ }
+
+ // open: bytes, size, and a usable modtime all come back.
+ r, size, modTime, err := store.open("blobid_123-abc")
+ if err != nil {
+ t.Fatalf("open: %v", err)
+ }
+ got, err := io.ReadAll(r)
+ r.Close()
+ if err != nil || !bytes.Equal(got, payload) {
+ t.Fatalf("round-trip mismatch: err=%v got=%q", err, got)
+ }
+ if size != int64(len(payload)) {
+ t.Fatalf("size = %d, want %d", size, len(payload))
+ }
+ if modTime.IsZero() {
+ t.Fatal("zero modtime")
+ }
+
+ // delete → open fails with the contractual not-exist error.
+ if err := store.remove("blobid_123-abc"); err != nil {
+ t.Fatalf("remove: %v", err)
+ }
+ if _, _, _, err := store.open("blobid_123-abc"); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("open after remove: want fs.ErrNotExist, got %v", err)
+ }
+ // GC unlinks best-effort, so removing an id that's already gone is not an
+ // error.
+ if err := store.remove("blobid_123-abc"); err != nil {
+ t.Fatalf("remove of missing id: %v", err)
+ }
+}
+
+// An aborted staged write leaves nothing fetchable — the id is never published.
+func TestDiskBlobStoreAbortDiscards(t *testing.T) {
+ store, err := newDiskBlobStore(filepath.Join(t.TempDir(), "blobs"))
+ if err != nil {
+ t.Fatalf("newDiskBlobStore: %v", err)
+ }
+ w, err := store.create()
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+ if _, err := w.Write([]byte("half an upload")); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ w.abort()
+ if _, _, _, err := store.open("neverpublished"); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("open of aborted write: want fs.ErrNotExist, got %v", err)
+ }
+}
diff --git a/db.go b/db.go
index 0797417..7859bbb 100644
--- a/db.go
+++ b/db.go
@@ -8,7 +8,6 @@ import (
"encoding/base64"
"encoding/hex"
"errors"
- "os"
"path/filepath"
"strings"
"time"
@@ -166,11 +165,11 @@ func (db *DB) setUpdatePin(v string) error { return db.setConfig("update_pin", v
// ciphertext blobs produced by the browser.
type DB struct {
sql *sql.DB
- // blobDir holds attachment/photo bytes as files (see putBlob/getBlob). It sits
- // next to the DB file (<db>-blobs/) so it's per-instance and a data snapshot
- // captures both by copying the DB and its sibling dir. Large media never bloats
- // the SQLite file itself (keeping WAL/VACUUM/snapshots lean).
- blobDir string
+ // blobs holds attachment/photo bytes (see putBlob / the upload handler).
+ // It's the blobStore seam: the default is on-disk files next to the DB
+ // (<db>-blobs/), but nothing outside blobstore.go may assume that — the
+ // blobs *table* keeps the metadata (size, thread tag) either way.
+ blobs blobStore
}
func openDB(path string) (*DB, error) {
@@ -190,10 +189,11 @@ func openDB(path string) (*DB, error) {
return nil, err
}
}
- db := &DB{sql: d, blobDir: strings.TrimSuffix(path, filepath.Ext(path)) + "-blobs"}
- if err := os.MkdirAll(db.blobDir, 0o755); err != nil {
+ blobs, err := newDiskBlobStore(strings.TrimSuffix(path, filepath.Ext(path)) + "-blobs")
+ if err != nil {
return nil, err
}
+ db := &DB{sql: d, blobs: blobs}
if err := db.migrate(); err != nil {
return nil, err
}
@@ -2356,15 +2356,10 @@ func (db *DB) messagesAfter(threadID string, after int64, limit int) ([]Message,
// the server stores only opaque bytes — no filename, type, or content. The
// referencing message (also encrypted) carries the metadata.
-// blobPath is the on-disk location of a disk-backed blob's bytes. Blob ids are
-// randToken (base64url), so they're always safe filename components — but the
-// HTTP handler still validates the id (validBlobID) before building this path.
-func (db *DB) blobPath(id string) string { return filepath.Join(db.blobDir, id) }
-
-// recordBlob inserts the metadata row for a disk-backed blob whose bytes the
-// caller has already written to blobPath(id). `data` is stored empty — the row
-// exists only for quota accounting and thread-scoped GC. threadID ties the blob
-// to its thread (empty → NULL, e.g. profile-avatar uploads that aren't
+// recordBlob inserts the metadata row for a store-backed blob whose bytes the
+// caller has already committed to db.blobs under id. `data` is stored empty —
+// the row exists only for quota accounting and thread-scoped GC. threadID ties
+// the blob to its thread (empty → NULL, e.g. profile-avatar uploads that aren't
// thread-scoped and are replaced rather than message-GC'd).
func (db *DB) recordBlob(id, threadID string, size int64) error {
var tid any
@@ -2378,23 +2373,31 @@ func (db *DB) recordBlob(id, threadID string, size int64) error {
return err
}
-// putBlob stores small in-memory blob bytes on disk + records the row. Kept for
-// callers that already hold the bytes (tests, any non-streaming path); the HTTP
-// upload path streams straight to disk instead (see uploadBlob).
+// putBlob stores small in-memory blob bytes in the store + records the row.
+// Kept for callers that already hold the bytes (tests, any non-streaming path);
+// the HTTP upload path streams through a staged write instead (see uploadBlob).
func (db *DB) putBlob(id, threadID string, data []byte) error {
- if err := os.WriteFile(db.blobPath(id), data, 0o644); err != nil {
+ w, err := db.blobs.create()
+ if err != nil {
+ return err
+ }
+ if _, err := w.Write(data); err != nil {
+ w.abort()
+ return err
+ }
+ if err := w.commit(id); err != nil {
return err
}
if err := db.recordBlob(id, threadID, int64(len(data))); err != nil {
- _ = os.Remove(db.blobPath(id))
+ _ = db.blobs.remove(id)
return err
}
return nil
}
// legacyBlob returns the bytes of a pre-disk (column-stored) blob, or nil if the
-// blob is disk-backed (empty column) or unknown. Only the HTTP handler's fallback
-// path calls this, after it fails to find the blob file.
+// blob is store-backed (empty column) or unknown. Only the HTTP handler's
+// fallback path calls this, after the store reports no such blob.
func (db *DB) legacyBlob(id string) ([]byte, error) {
var data []byte
err := db.sql.QueryRow("SELECT data FROM blobs WHERE id = ?", id).Scan(&data)
@@ -2420,8 +2423,8 @@ func (db *DB) blobBytesTotal() (int64, error) {
// deleteBlobs removes the given blobs *scoped to one thread*: a row only goes if
// its thread_id matches, so a client reporting ids to GC when it deletes its own
-// message can never drop another thread's (or another member's) blobs. On-disk
-// bytes are unlinked best-effort for each row actually removed.
+// message can never drop another thread's (or another member's) blobs. Stored
+// bytes are removed best-effort for each row actually removed.
func (db *DB) deleteBlobs(threadID string, ids []string) error {
for _, id := range ids {
if !validBlobID(id) {
@@ -2432,7 +2435,7 @@ func (db *DB) deleteBlobs(threadID string, ids []string) error {
return err
}
if n, _ := res.RowsAffected(); n > 0 {
- _ = os.Remove(db.blobPath(id))
+ _ = db.blobs.remove(id)
}
}
return nil
@@ -2462,7 +2465,7 @@ func (db *DB) deleteThreadBlobs(threadID string) error {
return err
}
for _, id := range ids {
- _ = os.Remove(db.blobPath(id))
+ _ = db.blobs.remove(id)
}
return nil
}
diff --git a/main.go b/main.go
index c4a5634..d751d74 100644
--- a/main.go
+++ b/main.go
@@ -1825,38 +1825,33 @@ func (s *Server) uploadBlob(w http.ResponseWriter, r *http.Request) {
httpError(w, http.StatusInternalServerError, "id error")
return
}
- // Stream to a temp file (never buffer a whole video in RAM); rename into place
- // only once it's fully written, so a partial/aborted upload is never served.
- // The "." in the temp name makes it an invalid blob id (validBlobID), so an
- // in-flight temp can never be fetched via /api/blob. Same dir as the final
- // file so the rename is atomic (same filesystem).
- tmp, err := os.CreateTemp(s.db.blobDir, "up.*.part")
+ // Stream into a staged store write (never buffer a whole video in RAM);
+ // commit under the id only once it's fully written, so a partial/aborted
+ // upload is never served.
+ blob, err := s.db.blobs.create()
if err != nil {
httpError(w, http.StatusInternalServerError, "storage error")
return
}
- tmpName := tmp.Name()
- size, err := io.Copy(tmp, r.Body)
- tmp.Close()
+ size, err := io.Copy(blob, r.Body)
if err != nil {
- os.Remove(tmpName)
+ blob.abort()
// MaxBytesReader trips here when the body exceeds the cap.
httpError(w, http.StatusRequestEntityTooLarge, "file too large")
return
}
if size == 0 {
- os.Remove(tmpName)
+ blob.abort()
httpError(w, http.StatusBadRequest, "empty upload")
return
}
// Re-check the quota against the real size now we have it.
if total, err := s.db.blobBytesTotal(); err == nil && total+size > s.blobQuota {
- os.Remove(tmpName)
+ blob.abort()
httpError(w, http.StatusInsufficientStorage, "storage full")
return
}
- if err := os.Rename(tmpName, s.db.blobPath(id)); err != nil {
- os.Remove(tmpName)
+ if err := blob.commit(id); err != nil {
httpError(w, http.StatusInternalServerError, "storage error")
return
}
@@ -1864,7 +1859,7 @@ func (s *Server) uploadBlob(w http.ResponseWriter, r *http.Request) {
// with its thread / when the referencing message is deleted. Optional: avatar
// uploads carry no thread.
if err := s.db.recordBlob(id, r.URL.Query().Get("thread"), size); err != nil {
- os.Remove(s.db.blobPath(id))
+ _ = s.db.blobs.remove(id)
httpError(w, http.StatusInternalServerError, "db error")
return
}
@@ -1886,15 +1881,13 @@ func (s *Server) getBlobHandler(w http.ResponseWriter, r *http.Request) {
// nosniff (set globally) stops a mistyped blob being sniffed as HTML; we serve
// everything as opaque bytes.
w.Header().Set("Content-Type", "application/octet-stream")
- // Disk-backed (all new blobs): stream from the file. ServeContent gives Range
- // support + conditional requests for free (matters for large video), and never
- // loads the whole file into RAM.
- if f, err := os.Open(s.db.blobPath(id)); err == nil {
+ // Store-backed (all new blobs): stream from the store. ServeContent gives
+ // Range support + conditional requests for free (matters for large video),
+ // and never loads the whole blob into RAM.
+ if f, _, modTime, err := s.db.blobs.open(id); err == nil {
defer f.Close()
- if fi, err := f.Stat(); err == nil {
- http.ServeContent(w, r, "", fi.ModTime(), f)
- return
- }
+ http.ServeContent(w, r, "", modTime, f)
+ return
}
// Legacy: bytes still in the DB column (photos uploaded before disk-backing).
data, err := s.db.legacyBlob(id)