| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "io" |
| 6 | "io/fs" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "time" |
| 10 | ) |
| 11 | |
| 12 | // blobStore is the seam between the messenger and wherever attachment bytes |
| 13 | // physically live. It exists so the flagship can later point media at an |
| 14 | // object store (issue #550) while self-hosters keep today's on-disk layout — |
| 15 | // the handlers and GC talk only to this interface. |
| 16 | // |
| 17 | // The store holds BYTES ONLY. Everything the server knows *about* a blob — |
| 18 | // its size (quota accounting) and thread tag (thread-scoped GC) — stays in |
| 19 | // SQLite's blobs table, which travels with the rest of the instance data. |
| 20 | // Content arrives as client-side ciphertext; the store never learns filenames, |
| 21 | // MIME types, or plaintext (those ride inside the encrypted message payload, |
| 22 | // per the E2EE rule). Objects are write-once and immutable: an id is minted |
| 23 | // exactly once (randToken), never rewritten, only ever deleted — so there is |
| 24 | // no versioning or CAS here, unlike home's vaultStore. |
| 25 | type blobStore interface { |
| 26 | // create begins a staged write. The staged bytes are invisible to open |
| 27 | // until commit publishes them under an id, so a partial or abandoned |
| 28 | // upload can never be served. The two-step shape mirrors what the upload |
| 29 | // handler actually does: it learns the size only after streaming the |
| 30 | // body, and must apply the quota verdict between "fully received" and |
| 31 | // "fetchable". |
| 32 | create() (blobWriter, error) |
| 33 | // open returns a reader over a committed blob's bytes, plus its size and |
| 34 | // modification time (http.ServeContent wants seek + modtime for Range and |
| 35 | // conditional requests — video playback scrubs). An id the store doesn't |
| 36 | // hold returns an error satisfying errors.Is(err, fs.ErrNotExist). |
| 37 | open(id string) (r io.ReadSeekCloser, size int64, modTime time.Time, err error) |
| 38 | // remove deletes a blob's bytes. Removing an id that isn't there is not |
| 39 | // an error — GC call sites unlink best-effort. |
| 40 | remove(id string) error |
| 41 | } |
| 42 | |
| 43 | // blobWriter is one staged write. Exactly one of commit or abort ends it. |
| 44 | type blobWriter interface { |
| 45 | io.Writer |
| 46 | // commit publishes the staged bytes under id: after commit, open(id) |
| 47 | // serves exactly what was written. Publication is atomic — a crash |
| 48 | // mid-commit leaves either nothing or the whole blob, never a readable |
| 49 | // partial. |
| 50 | commit(id string) error |
| 51 | // abort discards the staged bytes. Best-effort with no error to report: |
| 52 | // every caller is already unwinding a failure. |
| 53 | abort() |
| 54 | } |
| 55 | |
| 56 | // diskBlobStore is implementation #1: one file per blob id under a single |
| 57 | // directory. It wraps exactly the storage that predates the seam — the dir |
| 58 | // sits next to the DB file (<db>-blobs/) so it's per-instance and a data |
| 59 | // snapshot captures both by copying the DB and its sibling dir. Large media |
| 60 | // never bloats the SQLite file itself (keeping WAL/VACUUM/snapshots lean). |
| 61 | type diskBlobStore struct{ dir string } |
| 62 | |
| 63 | func newDiskBlobStore(dir string) (*diskBlobStore, error) { |
| 64 | if err := os.MkdirAll(dir, 0o755); err != nil { |
| 65 | return nil, err |
| 66 | } |
| 67 | return &diskBlobStore{dir: dir}, nil |
| 68 | } |
| 69 | |
| 70 | // path is the on-disk location of a blob's bytes. Blob ids are randToken |
| 71 | // (base64url), so they're always safe filename components — but the HTTP |
| 72 | // handler still validates the id (validBlobID) before it can reach here. |
| 73 | func (s *diskBlobStore) path(id string) string { return filepath.Join(s.dir, id) } |
| 74 | |
| 75 | func (s *diskBlobStore) create() (blobWriter, error) { |
| 76 | // The "." in the temp name makes it an invalid blob id (validBlobID), so |
| 77 | // an in-flight staged write can never be fetched via /api/blob. Same dir |
| 78 | // as the final file so the commit rename is atomic (same filesystem). |
| 79 | f, err := os.CreateTemp(s.dir, "up.*.part") |
| 80 | if err != nil { |
| 81 | return nil, err |
| 82 | } |
| 83 | return &diskBlobWriter{f: f, store: s}, nil |
| 84 | } |
| 85 | |
| 86 | func (s *diskBlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, error) { |
| 87 | f, err := os.Open(s.path(id)) |
| 88 | if err != nil { |
| 89 | return nil, 0, time.Time{}, err // wraps fs.ErrNotExist for a missing id |
| 90 | } |
| 91 | fi, err := f.Stat() |
| 92 | if err != nil { |
| 93 | f.Close() |
| 94 | return nil, 0, time.Time{}, err |
| 95 | } |
| 96 | return f, fi.Size(), fi.ModTime(), nil |
| 97 | } |
| 98 | |
| 99 | func (s *diskBlobStore) remove(id string) error { |
| 100 | if err := os.Remove(s.path(id)); err != nil && !errors.Is(err, fs.ErrNotExist) { |
| 101 | return err |
| 102 | } |
| 103 | return nil |
| 104 | } |
| 105 | |
| 106 | type diskBlobWriter struct { |
| 107 | f *os.File |
| 108 | store *diskBlobStore |
| 109 | } |
| 110 | |
| 111 | func (w *diskBlobWriter) Write(p []byte) (int, error) { return w.f.Write(p) } |
| 112 | |
| 113 | func (w *diskBlobWriter) commit(id string) error { |
| 114 | // Close error deliberately ignored, matching the pre-seam upload path: |
| 115 | // the copy's own error return is the signal callers act on. |
| 116 | w.f.Close() |
| 117 | if err := os.Rename(w.f.Name(), w.store.path(id)); err != nil { |
| 118 | os.Remove(w.f.Name()) |
| 119 | return err |
| 120 | } |
| 121 | return nil |
| 122 | } |
| 123 | |
| 124 | func (w *diskBlobWriter) abort() { |
| 125 | w.f.Close() |
| 126 | os.Remove(w.f.Name()) |
| 127 | } |
| 128 | |
amadan