1package blob23import (4 "context"5 "crypto/sha256"6 "database/sql"7 "encoding/hex"8 "errors"9 "fmt"10 "io"11 "os"12 "path/filepath"13 "strconv"14 "strings"15 "time"1617 "github.com/FelineStateMachine/tinyrelay/internal/storage"18)1920// Put stores bytes using their SHA-256 digest as the durable filename.21func (s *Service) Put(ctx context.Context, options PutOptions) (Blob, error) {22 if options.Reader == nil {23 return Blob{}, errors.New("blob reader is required")24 }25 entry, _, err := s.putOptions(ctx, options)26 return entry, err27}2829func (s *Service) putOptions(ctx context.Context, options PutOptions) (Blob, bool, error) {30 return s.putValidatedWithMetadata(ctx, options.Reader, options.Type, options.Uploader, options.Hash, options.Name, options.Path, options.Access, options.Purpose, nil, options.Commit)31}3233// Get opens a verified metadata-backed blob for streaming to another service.34func (s *Service) Get(ctx context.Context, sha string) (Blob, io.ReadCloser, error) {35 if !shaPattern.MatchString(sha) {36 return Blob{}, nil, errors.New("invalid: bad blob sha256")37 }38 entry, err := s.lookup(ctx, sha)39 if err != nil {40 return Blob{}, nil, err41 }42 if s.blocked(ctx, sha) {43 return Blob{}, nil, ErrBlocked44 }45 file, err := os.Open(filepath.Join(s.root, sha))46 if err != nil {47 return Blob{}, nil, err48 }49 return entry, file, nil50}5152// ListMetadata returns all blobs uploaded by the supplied pubkey.53func (s *Service) ListMetadata(ctx context.Context, uploader string) ([]Blob, error) {54 rows, err := s.store.DB().QueryContext(ctx, `SELECT b.sha256,b.size,b.type,c.uploader,b.uploaded55 FROM blobs b JOIN blob_claims c ON c.sha256=b.sha256 WHERE c.uploader=?56 ORDER BY b.uploaded DESC,b.sha256 DESC`, uploader)57 if err != nil {58 return nil, fmt.Errorf("list blob metadata: %w", err)59 }60 defer rows.Close()61 entries := make([]Blob, 0)62 for rows.Next() {63 var entry Blob64 if err := rows.Scan(&entry.SHA256, &entry.Size, &entry.Type, &entry.Uploader, &entry.Uploaded); err != nil {65 return nil, fmt.Errorf("scan blob metadata: %w", err)66 }67 entries = append(entries, entry)68 }69 return entries, rows.Err()70}7172// ListPage returns a BUD-12 cursor page. The cursor is the last blob hash73// returned by the previous page and is intentionally opaque to callers.74func (s *Service) ListPage(ctx context.Context, uploader string, limit int, cursor string) ([]Blob, string, error) {75 if limit < 1 {76 limit = 2077 }78 if limit > 100 {79 limit = 10080 }81 query := `SELECT b.sha256,b.size,b.type,c.uploader,b.uploaded FROM blobs b82 JOIN blob_claims c ON c.sha256=b.sha256 WHERE c.uploader=?`83 args := []any{uploader}84 if cursor != "" {85 var uploaded int6486 if err := s.store.DB().QueryRowContext(ctx, `SELECT b.uploaded FROM blobs b87 JOIN blob_claims c ON c.sha256=b.sha256 WHERE b.sha256=? AND c.uploader=?`, cursor, uploader).Scan(&uploaded); err != nil {88 return nil, "", fmt.Errorf("resolve blob cursor: %w", err)89 }90 query += " AND (b.uploaded < ? OR (b.uploaded = ? AND b.sha256 < ?))"91 args = append(args, uploaded, uploaded, cursor)92 }93 query += " ORDER BY b.uploaded DESC,b.sha256 DESC LIMIT ?"94 args = append(args, limit+1)95 rows, err := s.store.DB().QueryContext(ctx, query, args...)96 if err != nil {97 return nil, "", fmt.Errorf("list blob page: %w", err)98 }99 defer rows.Close()100 entries := make([]Blob, 0, limit)101 for rows.Next() {102 var entry Blob103 if err := rows.Scan(&entry.SHA256, &entry.Size, &entry.Type, &entry.Uploader, &entry.Uploaded); err != nil {104 return nil, "", fmt.Errorf("scan blob page: %w", err)105 }106 entries = append(entries, entry)107 }108 if err := rows.Err(); err != nil {109 return nil, "", fmt.Errorf("read blob page: %w", err)110 }111 next := ""112 if len(entries) > limit {113 next = entries[limit-1].SHA256114 entries = entries[:limit]115 }116 return entries, next, nil117}118119// Delete removes a blob's metadata and content. Authorization is the caller's responsibility.120func (s *Service) Delete(ctx context.Context, sha string) error {121 if !shaPattern.MatchString(sha) {122 return errors.New("invalid: bad blob sha256")123 }124 if err := s.store.WithTx(ctx, func(tx *sql.Tx) error {125 if _, err := tx.ExecContext(ctx, "INSERT INTO blob_tombstones(sha256,deleted_at) VALUES(?,?) ON CONFLICT(sha256) DO UPDATE SET deleted_at=excluded.deleted_at", sha, time.Now().UTC().Unix()); err != nil {126 return fmt.Errorf("record blob deletion: %w", err)127 }128 if _, err := tx.ExecContext(ctx, "DELETE FROM blobs WHERE sha256=?", sha); err != nil {129 return fmt.Errorf("delete blob metadata: %w", err)130 }131 if _, err := tx.ExecContext(ctx, "DELETE FROM blob_claim_metadata WHERE sha256=?", sha); err != nil {132 return fmt.Errorf("delete blob claim metadata: %w", err)133 }134 return nil135 }); err != nil {136 return err137 }138 if err := os.Remove(filepath.Join(s.root, sha)); err != nil && !errors.Is(err, os.ErrNotExist) {139 return fmt.Errorf("delete blob content: %w", err)140 }141 return nil142}143144// DeleteForUploader removes one owner's claim. The physical object remains145// while another owner still references it, which makes deduplicated uploads146// safe to manage independently.147func (s *Service) DeleteForUploader(ctx context.Context, sha, uploader string) error {148 if !shaPattern.MatchString(sha) {149 return errors.New("invalid: bad blob sha256")150 }151 removed := false152 if err := s.store.WithTx(ctx, func(tx *sql.Tx) error {153 result, err := tx.ExecContext(ctx, "DELETE FROM blob_claims WHERE sha256=? AND uploader=?", sha, uploader)154 if err != nil {155 return fmt.Errorf("delete blob claim: %w", err)156 }157 changed, err := result.RowsAffected()158 if err != nil {159 return fmt.Errorf("check blob claim deletion: %w", err)160 }161 if changed == 0 {162 return sql.ErrNoRows163 }164 if _, err := tx.ExecContext(ctx, "DELETE FROM blob_claim_metadata WHERE sha256=? AND uploader=?", sha, uploader); err != nil {165 return fmt.Errorf("delete blob claim metadata: %w", err)166 }167 var remaining int168 if err := tx.QueryRowContext(ctx, "SELECT count(*) FROM blob_claims WHERE sha256=?", sha).Scan(&remaining); err != nil {169 return fmt.Errorf("count blob claims: %w", err)170 }171 if remaining == 0 {172 removed = true173 if _, err := tx.ExecContext(ctx, "INSERT INTO blob_tombstones(sha256,deleted_at) VALUES(?,?) ON CONFLICT(sha256) DO UPDATE SET deleted_at=excluded.deleted_at", sha, time.Now().UTC().Unix()); err != nil {174 return fmt.Errorf("record blob deletion: %w", err)175 }176 if _, err := tx.ExecContext(ctx, "DELETE FROM blobs WHERE sha256=?", sha); err != nil {177 return fmt.Errorf("delete blob metadata: %w", err)178 }179 }180 return nil181 }); err != nil {182 return err183 }184 if removed {185 if err := os.Remove(filepath.Join(s.root, sha)); err != nil && !errors.Is(err, os.ErrNotExist) {186 return fmt.Errorf("delete blob content: %w", err)187 }188 }189 return nil190}191192// Descriptor returns the Blossom descriptor and NIP-94 tags for a blob.193func (s *Service) Descriptor(baseURL string, entry Blob) map[string]any {194 return map[string]any{195 "url": baseURL + "/" + entry.SHA256 + extensionSuffix(entry.Type),196 "sha256": entry.SHA256,197 "size": entry.Size,198 "type": entry.Type,199 "uploaded": entry.Uploaded,200 "nip94": [][]string{{"url", baseURL + "/" + entry.SHA256 + extensionSuffix(entry.Type)}, {"m", entry.Type}, {"x", entry.SHA256}, {"ox", entry.SHA256}, {"size", strconv.FormatInt(entry.Size, 10)}},201 }202}203func (s *Service) put(ctx context.Context, source io.Reader, typ, uploader, claimed string) (Blob, bool, error) {204 return s.putValidatedWithCommit(ctx, source, typ, uploader, claimed, nil, nil)205}206207func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploader, claimed string, validate func(string) error) (Blob, bool, error) {208 return s.putValidatedWithMetadata(ctx, source, typ, uploader, claimed, "", "", "", "", validate, nil)209}210211func (s *Service) putValidatedWithAccess(ctx context.Context, source io.Reader, typ, uploader, claimed, access string, validate func(string) error) (Blob, bool, error) {212 return s.putValidatedWithMetadata(ctx, source, typ, uploader, claimed, "", "", access, "", validate, nil)213}214215func (s *Service) putValidatedWithCommit(ctx context.Context, source io.Reader, typ, uploader, claimed string, validate func(string) error, commit func(context.Context, *sql.Tx, Blob, bool) error) (Blob, bool, error) {216 return s.putValidatedWithMetadata(ctx, source, typ, uploader, claimed, "", "", "", "", validate, commit)217}218219func (s *Service) putValidatedWithMetadata(ctx context.Context, source io.Reader, typ, uploader, claimed, name, path, access, purpose string, validate func(string) error, commit func(context.Context, *sql.Tx, Blob, bool) error) (Blob, bool, error) {220 var err error221 access, err = parseAccess(access)222 if err != nil {223 return Blob{}, false, err224 }225 name, path, err = claimMetadata(name, path)226 if err != nil {227 return Blob{}, false, err228 }229 limits := s.currentLimits()230 streamLimit := limits.MaxFileBytes231 quotaStream := false232 if streamLimit == 0 && limits.UserStorageBytes > 0 && uploader != "" {233 streamLimit = limits.UserStorageBytes234 quotaStream = true235 }236 if streamLimit > 0 && streamLimit < int64(^uint64(0)>>1) {237 source = io.LimitReader(source, streamLimit+1)238 }239 tmp, err := os.CreateTemp(s.root, ".upload-")240 if err != nil {241 return Blob{}, false, fmt.Errorf("create upload temporary file: %w", err)242 }243 tmpName := tmp.Name()244 defer os.Remove(tmpName)245 hasher := sha256.New()246 count, err := io.Copy(io.MultiWriter(tmp, hasher), source)247 if err != nil {248 _ = tmp.Close()249 return Blob{}, false, fmt.Errorf("read upload: %w", err)250 }251 if limits.MaxFileBytes > 0 && count > limits.MaxFileBytes {252 _ = tmp.Close()253 return Blob{}, false, ErrFileTooLarge254 }255 if quotaStream && count > streamLimit {256 _ = tmp.Close()257 return Blob{}, false, ErrQuotaExceeded258 }259 if err := tmp.Sync(); err != nil {260 _ = tmp.Close()261 return Blob{}, false, fmt.Errorf("sync upload: %w", err)262 }263 if err := tmp.Close(); err != nil {264 return Blob{}, false, fmt.Errorf("close upload: %w", err)265 }266 sha := hex.EncodeToString(hasher.Sum(nil))267 if claimed != "" && claimed != sha {268 return Blob{}, false, ErrHashMismatch269 }270 if validate != nil {271 if err := validate(sha); err != nil {272 return Blob{}, false, err273 }274 }275 return s.installUpload(ctx, uploadCandidate{path: tmpName, hash: sha, size: count, typ: typ, uploader: uploader, name: name, metadataPath: path, access: access, purpose: purpose, commit: commit})276}277278type uploadCandidate struct {279 path, hash, typ, uploader, reservation, name, metadataPath, access, purpose string280 size int64281 commit func(context.Context, *sql.Tx, Blob, bool) error282}283284func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) (Blob, bool, error) {285 sha, count, typ, uploader := candidate.hash, candidate.size, candidate.typ, candidate.uploader286 access, accessErr := parseAccess(candidate.access)287 if accessErr != nil {288 return Blob{}, false, accessErr289 }290 // Serialize the lookup, quota check, installation and metadata transaction.291 // This closes the race where concurrent uploads could each observe the same292 // remaining quota before one of them installs a new physical object.293 s.quotaMu.Lock()294 defer s.quotaMu.Unlock()295 if s.blocked(ctx, sha) {296 return Blob{}, false, ErrBlocked297 }298 terms, err := s.uploadTerms(ctx, uploader)299 if err != nil {300 return Blob{}, false, err301 }302 limits := s.currentLimits()303 if limits.MaxFileBytes > 0 && count > limits.MaxFileBytes {304 return Blob{}, false, ErrFileTooLarge305 }306 if limits.UserStorageBytes > 0 && uploader != "" && count > limits.UserStorageBytes {307 return Blob{}, false, ErrQuotaExceeded308 }309 if existing, err := s.lookup(ctx, sha); err == nil {310 existingAccess, accessErr := s.BlobAccess(ctx, sha)311 if accessErr != nil {312 return Blob{}, false, accessErr313 }314 if existingAccess != access {315 return Blob{}, false, ErrAccessConflict316 }317 if terms.Encrypted && !encryptedTypes[existing.Type] {318 return Blob{}, false, ErrNotEncrypted319 }320 if uploader != "" {321 if limits.UserStorageBytes > 0 && !s.hasClaim(ctx, sha, uploader) {322 used, usageErr := s.quotaUsageExcept(ctx, uploader, candidate.reservation)323 if usageErr != nil {324 return Blob{}, false, fmt.Errorf("check uploader storage quota: %w", usageErr)325 }326 if existing.Size > limits.UserStorageBytes-used {327 return Blob{}, false, ErrQuotaExceeded328 }329 }330 }331 if err := s.store.WithTx(ctx, func(tx *sql.Tx) error {332 if candidate.commit != nil {333 if err := candidate.commit(ctx, tx, existing, false); err != nil {334 return err335 }336 }337 if uploader != "" {338 if _, err := tx.ExecContext(ctx, claimSQL, sha, uploader, time.Now().Unix(), terms.ExpiresAt); err != nil {339 return err340 }341 if err := s.saveClaimMetadata(ctx, tx, sha, uploader, candidate.name, candidate.metadataPath, candidate.purpose); err != nil {342 return err343 }344 }345 if candidate.reservation != "" {346 if _, err := tx.ExecContext(ctx, "DELETE FROM multipart_uploads WHERE id=?", candidate.reservation); err != nil {347 return err348 }349 }350 return nil351 }); err != nil {352 return Blob{}, false, err353 }354 // A crash after a scoped metadata commit may leave the bytes missing.355 // A retry supplies those same verified bytes without losing the scope.356 if _, err := os.Stat(filepath.Join(s.root, sha)); errors.Is(err, os.ErrNotExist) {357 if err := s.installBlobFile(candidate.path, sha); err != nil {358 return Blob{}, false, err359 }360 } else if err != nil {361 return Blob{}, false, err362 }363 return existing, false, nil364 } else if !errors.Is(err, sql.ErrNoRows) {365 return Blob{}, false, err366 }367 if limits.UserStorageBytes > 0 && uploader != "" {368 used, err := s.quotaUsageExcept(ctx, uploader, candidate.reservation)369 if err != nil {370 return Blob{}, false, fmt.Errorf("check uploader storage quota: %w", err)371 }372 if count > limits.UserStorageBytes-used {373 return Blob{}, false, ErrQuotaExceeded374 }375 }376 // The type is settled before the bytes are installed so a plain upload377 // under an encrypted-only grant leaves nothing behind. A declared octet378 // stream is sniffed: encrypted bytes stay one, a page or an image does not.379 typ = contentType(typ)380 if typ == "application/octet-stream" {381 typ = detectFileType(candidate.path)382 }383 if terms.Encrypted && !encryptedTypes[typ] {384 return Blob{}, false, ErrNotEncrypted385 }386 stage := ""387 if candidate.commit != nil {388 stage = filepath.Join(s.root, ".scoped-"+sha)389 if err := os.Rename(candidate.path, stage); err != nil {390 return Blob{}, false, fmt.Errorf("stage scoped blob: %w", err)391 }392 if err := syncDirectory(s.root); err != nil {393 return Blob{}, false, fmt.Errorf("sync staged blob directory: %w", err)394 }395 }396 if candidate.commit == nil {397 if err := s.installBlobFile(candidate.path, sha); err != nil {398 return Blob{}, false, err399 }400 }401 now := time.Now().UTC().Unix()402 entry := Blob{SHA256: sha, Size: count, Type: typ, Uploader: uploader, Uploaded: now}403 err = s.store.WithTx(ctx, func(tx *sql.Tx) error {404 if _, err := tx.ExecContext(ctx, "DELETE FROM blob_tombstones WHERE sha256=?", entry.SHA256); err != nil {405 return fmt.Errorf("clear blob deletion tombstone: %w", err)406 }407 if _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO blobs(sha256,size,type,uploader,uploaded,access) VALUES(?,?,?,?,?,?)", entry.SHA256, entry.Size, entry.Type, entry.Uploader, entry.Uploaded, access); err != nil {408 return fmt.Errorf("record blob: %w", err)409 }410 if entry.Uploader != "" {411 if _, err := tx.ExecContext(ctx, claimSQL, entry.SHA256, entry.Uploader, entry.Uploaded, terms.ExpiresAt); err != nil {412 return fmt.Errorf("record blob claim: %w", err)413 }414 if err := s.saveClaimMetadata(ctx, tx, entry.SHA256, entry.Uploader, candidate.name, candidate.metadataPath, candidate.purpose); err != nil {415 return err416 }417 }418 if candidate.reservation != "" {419 if _, err := tx.ExecContext(ctx, "DELETE FROM multipart_uploads WHERE id=?", candidate.reservation); err != nil {420 return fmt.Errorf("release upload reservation: %w", err)421 }422 }423 if candidate.commit != nil {424 if err := candidate.commit(ctx, tx, entry, true); err != nil {425 return err426 }427 }428 return nil429 })430 if err != nil {431 if stage != "" {432 _ = os.Remove(stage)433 }434 return Blob{}, false, err435 }436 if candidate.commit != nil {437 if err := s.installBlobFile(stage, sha); err != nil {438 return Blob{}, false, err439 }440 }441 if saved, err := s.lookup(ctx, sha); err == nil {442 return saved, saved.Uploader == uploader && saved.Uploaded == now, nil443 }444 return entry, true, nil445}446447func (s *Service) installBlobFile(source, hash string) error {448 if err := os.Rename(source, filepath.Join(s.root, hash)); err != nil {449 return fmt.Errorf("install blob: %w", err)450 }451 if err := syncDirectory(s.root); err != nil {452 return fmt.Errorf("sync blob directory: %w", err)453 }454 return nil455}456457func (s *Service) quotaUsage(ctx context.Context, uploader string) (int64, error) {458 return s.quotaUsageExcept(ctx, uploader, "")459}460461func (s *Service) quotaUsageExcept(ctx context.Context, uploader, reservation string) (int64, error) {462 var used sql.NullInt64463 err := s.store.DB().QueryRowContext(ctx, `SELECT COALESCE((SELECT SUM(b.size)464 FROM blobs b JOIN blob_claims c ON c.sha256=b.sha256 WHERE c.uploader=?),0) +465 COALESCE((SELECT SUM(length) FROM multipart_uploads WHERE uploader=? AND id != ?),0)`, uploader, uploader, reservation).Scan(&used)466 if err != nil {467 return 0, err468 }469 return used.Int64, nil470}471472// claimSQL records a claim. The insert trigger on blobs has usually made the473// row already, without an expiry, so the newest upload's terms replace it.474const claimSQL = "INSERT INTO blob_claims(sha256,uploader,claimed_at,expires) VALUES(?,?,?,?) ON CONFLICT(sha256,uploader) DO UPDATE SET expires=excluded.expires"475476func (s *Service) claim(ctx context.Context, sha, uploader string, expires int64) error {477 _, err := s.store.DB().ExecContext(ctx, claimSQL, sha, uploader, time.Now().UTC().Unix(), expires)478 if err != nil {479 return fmt.Errorf("record blob claim: %w", err)480 }481 return nil482}483484func (s *Service) uploadTerms(ctx context.Context, uploader string) (UploadTerms, error) {485 if s.config.UploadTerms == nil || uploader == "" {486 return UploadTerms{}, nil487 }488 // A refusal keeps its reason so the door answers with the right status.489 return s.config.UploadTerms(ctx, uploader)490}491492// SweepExpired removes the blobs whose every claim has lapsed and releases493// lapsed claims on blobs someone else still holds, so an agent's files leave494// with its site while a person's claim keeps them. It returns how many blobs495// were removed.496func (s *Service) SweepExpired(ctx context.Context, now int64) (int, error) {497 s.quotaMu.Lock()498 defer s.quotaMu.Unlock()499 rows, err := s.store.DB().QueryContext(ctx, `SELECT DISTINCT sha256 FROM blob_claims c WHERE c.expires>0 AND c.expires<=?500 AND NOT EXISTS(SELECT 1 FROM blob_claims live WHERE live.sha256=c.sha256 AND (live.expires=0 OR live.expires>?))`, now, now)501 if err != nil {502 return 0, fmt.Errorf("list expired blobs: %w", err)503 }504 var expired []string505 for rows.Next() {506 var sha string507 if err := rows.Scan(&sha); err != nil {508 _ = rows.Close()509 return 0, err510 }511 expired = append(expired, sha)512 }513 if err := errors.Join(rows.Err(), rows.Close()); err != nil {514 return 0, err515 }516 for _, sha := range expired {517 if err := s.Delete(ctx, sha); err != nil {518 return 0, err519 }520 }521 if _, err := s.store.DB().ExecContext(ctx, "DELETE FROM blob_claims WHERE expires>0 AND expires<=?", now); err != nil {522 return 0, fmt.Errorf("release expired blob claims: %w", err)523 }524 return len(expired), nil525}526527func (s *Service) currentLimits() Limits {528 if s.config.Limits == nil {529 return Limits{}530 }531 limits := s.config.Limits()532 if limits.MaxFileBytes < 0 {533 limits.MaxFileBytes = 0534 }535 if limits.UserStorageBytes < 0 {536 limits.UserStorageBytes = 0537 }538 return limits539}540541func (s *Service) hasClaim(ctx context.Context, sha, uploader string) bool {542 var found int543 err := s.store.DB().QueryRowContext(ctx, "SELECT 1 FROM blob_claims WHERE sha256=? AND uploader=?", sha, uploader).Scan(&found)544 return err == nil && found == 1545}546547func (s *Service) lookup(ctx context.Context, sha string) (Blob, error) {548 var entry Blob549 err := s.store.DB().QueryRowContext(ctx, "SELECT sha256,size,type,uploader,uploaded FROM blobs WHERE sha256=?", sha).Scan(&entry.SHA256, &entry.Size, &entry.Type, &entry.Uploader, &entry.Uploaded)550 return entry, err551}552553// BlobAccess returns the durable retrieval policy for a blob. Older databases554// are treated as public through the column default and NULL-safe fallback.555func (s *Service) BlobAccess(ctx context.Context, sha string) (string, error) {556 var access string557 err := s.store.DB().QueryRowContext(ctx, "SELECT COALESCE(access,?) FROM blobs WHERE sha256=?", AccessPublic, sha).Scan(&access)558 if err != nil {559 return "", err560 }561 parsed, parseErr := parseAccess(access)562 if parseErr != nil {563 return "", parseErr564 }565 return parsed, nil566}567568func parseAccess(access string) (string, error) {569 access = strings.ToLower(strings.TrimSpace(access))570 if access == "" || access == AccessPublic {571 return AccessPublic, nil572 }573 if access == AccessMembers {574 return AccessMembers, nil575 }576 return "", ErrInvalidAccess577}578579func (s *Service) blocked(ctx context.Context, sha string) bool {580 var found int581 err := s.store.DB().QueryRowContext(ctx, `SELECT 1 FROM hidden_events WHERE id=?582 UNION ALL SELECT 1 FROM blob_blocks WHERE sha256=? LIMIT 1`, sha, sha).Scan(&found)583 return err == nil && found == 1584}585586func reconcile(ctx context.Context, store *storage.Store, root string) error {587 entries, err := os.ReadDir(root)588 if err != nil {589 return fmt.Errorf("scan blob directory: %w", err)590 }591 for _, entry := range entries {592 if entry.IsDir() {593 continue594 }595 name := entry.Name()596 if strings.HasPrefix(name, ".scoped-") {597 sha := strings.TrimPrefix(name, ".scoped-")598 if !shaPattern.MatchString(sha) {599 if err := os.Remove(filepath.Join(root, name)); err != nil {600 return fmt.Errorf("remove invalid staged blob: %w", err)601 }602 continue603 }604 if err := reconcileStaged(ctx, store, root, name, sha); err != nil {605 return err606 }607 continue608 }609 if strings.HasPrefix(name, ".upload-") || strings.HasPrefix(name, ".multipart-chunk-") {610 if err := os.Remove(filepath.Join(root, name)); err != nil {611 return fmt.Errorf("remove abandoned upload: %w", err)612 }613 continue614 }615 if !shaPattern.MatchString(name) {616 continue617 }618 var deleted int619 if err := store.DB().QueryRowContext(ctx, "SELECT 1 FROM blob_tombstones WHERE sha256=? UNION ALL SELECT 1 FROM blob_blocks WHERE sha256=? LIMIT 1", name, name).Scan(&deleted); err == nil {620 continue621 } else if !errors.Is(err, sql.ErrNoRows) {622 return fmt.Errorf("check deleted blob %q: %w", name, err)623 }624 var known int625 if err := store.DB().QueryRowContext(ctx, "SELECT 1 FROM blobs WHERE sha256=?", name).Scan(&known); err == nil {626 continue627 } else if !errors.Is(err, sql.ErrNoRows) {628 return fmt.Errorf("check orphan blob %q: %w", name, err)629 }630 if err := recoverBlob(ctx, store, root, name); err != nil {631 return err632 }633 }634 return nil635}636637func reconcileStaged(ctx context.Context, store *storage.Store, root, name, sha string) error {638 var size int64639 if err := store.DB().QueryRowContext(ctx, "SELECT size FROM blobs WHERE sha256=?", sha).Scan(&size); errors.Is(err, sql.ErrNoRows) {640 if removeErr := os.Remove(filepath.Join(root, name)); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) {641 return removeErr642 }643 return nil644 } else if err != nil {645 return fmt.Errorf("check staged blob %q: %w", sha, err)646 }647 file, err := os.Open(filepath.Join(root, name))648 if err != nil {649 return err650 }651 info, err := file.Stat()652 if err != nil {653 _ = file.Close()654 return err655 }656 if info.Size() != size {657 _ = file.Close()658 return fmt.Errorf("staged blob %q has size %d, want %d", sha, info.Size(), size)659 }660 hasher := sha256.New()661 if _, err := io.Copy(hasher, file); err != nil {662 _ = file.Close()663 return err664 }665 if err := file.Close(); err != nil {666 return err667 }668 if hex.EncodeToString(hasher.Sum(nil)) != sha {669 return fmt.Errorf("staged blob %q failed content hash", sha)670 }671 if _, err := os.Stat(filepath.Join(root, sha)); errors.Is(err, os.ErrNotExist) {672 if err := os.Rename(filepath.Join(root, name), filepath.Join(root, sha)); err != nil {673 return err674 }675 } else if err != nil {676 return err677 } else {678 _ = os.Remove(filepath.Join(root, name))679 }680 return syncDirectory(root)681}682683func recoverBlob(ctx context.Context, store *storage.Store, root, name string) error {684 file, err := os.Open(filepath.Join(root, name))685 if err != nil {686 return fmt.Errorf("open orphan blob %q: %w", name, err)687 }688 hash := sha256.New()689 size, copyErr := io.Copy(hash, file)690 closeErr := file.Close()691 if copyErr != nil || closeErr != nil {692 return fmt.Errorf("read orphan blob %q: %w", name, errors.Join(copyErr, closeErr))693 }694 if hex.EncodeToString(hash.Sum(nil)) != name {695 return fmt.Errorf("orphan blob %q failed content hash", name)696 }697 info, err := os.Stat(filepath.Join(root, name))698 if err != nil {699 return fmt.Errorf("stat orphan blob %q: %w", name, err)700 }701 _, err = store.DB().ExecContext(ctx, "INSERT OR IGNORE INTO blobs(sha256,size,type,uploader,uploaded) VALUES(?,?,?,?,?)", name, size, "application/octet-stream", "", info.ModTime().Unix())702 if err != nil {703 return fmt.Errorf("recover orphan blob %q: %w", name, err)704 }705 return nil706}707708func syncDirectory(path string) error {709 directory, err := os.Open(path)710 if err != nil {711 return err712 }713 defer directory.Close()714 return directory.Sync()715}