// Package blob serves directory media from cloud storage, held fully in memory with stored thumbnails. package blob import ( "bytes" "context" "errors" "fmt" "image" "image/jpeg" "io" "log" "net/http" "path" "strings" "sync" "time" _ "image/gif" _ "image/png" "github.com/rwcarlsen/goexif/exif" "golang.org/x/image/draw" _ "golang.org/x/image/webp" "cloud.google.com/go/storage" "google.golang.org/api/iterator" ) const ( Bucket = "heliosian-media" refreshInterval = 5 * time.Minute thumbWidth = 480 thumbSuffix = "-thumb" thumbExt = ".jpg" thumbMime = "image/jpeg" fetchWorkers = 32 ) var folders = []string{"people", "families"} type entry struct { name string generation int64 mimeType string data []byte thumb []byte } type object struct { name string generation int64 mimeType string } type Store struct { bucket *storage.BucketHandle mu sync.RWMutex entries map[string]*entry } func New() (*Store, error) { client, err := storage.NewClient(context.Background()) if err != nil { return nil, fmt.Errorf("storage client: %w", err) } s := &Store{bucket: client.Bucket(Bucket), entries: map[string]*entry{}} if err := s.refresh(); err != nil { return nil, err } go s.refreshLoop() return s, nil } func Register(mux *http.ServeMux, s *Store) { mux.HandleFunc("GET /blob/{folder}/{name}", s.serve) } func (s *Store) refreshLoop() { for range time.Tick(refreshInterval) { if err := s.refresh(); err != nil { log.Printf("[ERROR] blob refresh: %v", err) } } } func (s *Store) refresh() error { start := time.Now() ctx := context.Background() primaries := map[string]object{} thumbs := map[string]object{} for _, folder := range folders { it := s.bucket.Objects(ctx, &storage.Query{Prefix: folder + "/"}) for { attrs, err := it.Next() if errors.Is(err, iterator.Done) { break } if err != nil { return fmt.Errorf("list %s: %w", folder, err) } base := strings.TrimSuffix(path.Base(attrs.Name), path.Ext(attrs.Name)) o := object{name: attrs.Name, generation: attrs.Generation, mimeType: attrs.ContentType} if strings.HasSuffix(base, thumbSuffix) { thumbs[folder+"/"+strings.TrimSuffix(base, thumbSuffix)] = o continue } key := folder + "/" + base if existing, ok := primaries[key]; ok { return fmt.Errorf("duplicate media for %s: %s and %s", key, existing.name, attrs.Name) } primaries[key] = o } } s.mu.RLock() missing := []string{} for key, o := range primaries { if cached, ok := s.entries[key]; !ok || cached.generation != o.generation { missing = append(missing, key) } } s.mu.RUnlock() fetched := map[string]*entry{} var mu sync.Mutex var firstErr error var wg sync.WaitGroup slots := make(chan struct{}, fetchWorkers) for _, key := range missing { wg.Add(1) slots <- struct{}{} go func() { defer wg.Done() defer func() { <-slots }() e, err := s.load(ctx, primaries[key], thumbs[key]) mu.Lock() defer mu.Unlock() if err != nil { if firstErr == nil { firstErr = err } return } fetched[key] = e }() } wg.Wait() if firstErr != nil { return firstErr } next := make(map[string]*entry, len(primaries)) var totalBytes int64 s.mu.Lock() for key, o := range primaries { if e, ok := fetched[key]; ok { next[key] = e } else if cached, ok := s.entries[key]; ok && cached.generation == o.generation { next[key] = cached } } s.entries = next for _, e := range next { totalBytes += int64(len(e.data) + len(e.thumb)) } s.mu.Unlock() log.Printf("blob store: %d files, %d fetched, %.1f MB in memory in %s", len(next), len(fetched), float64(totalBytes)/1e6, time.Since(start).Round(time.Millisecond)) return nil } func (s *Store) load(ctx context.Context, primary, thumb object) (*entry, error) { data, err := s.read(ctx, primary.name) if err != nil { return nil, err } e := &entry{name: primary.name, generation: primary.generation, mimeType: primary.mimeType, data: data} if !strings.HasPrefix(primary.mimeType, "image/") { return e, nil } if thumb.name == "" { return nil, fmt.Errorf("no thumbnail stored for %s", primary.name) } e.thumb, err = s.read(ctx, thumb.name) if err != nil { return nil, err } return e, nil } func (s *Store) read(ctx context.Context, name string) ([]byte, error) { r, err := s.bucket.Object(name).NewReader(ctx) if err != nil { return nil, fmt.Errorf("read %s: %w", name, err) } defer r.Close() content, err := io.ReadAll(r) if err != nil { return nil, fmt.Errorf("read %s: %w", name, err) } return content, nil } func (s *Store) write(ctx context.Context, name, mimeType string, content []byte) error { w := s.bucket.Object(name).NewWriter(ctx) w.ContentType = mimeType if _, err := w.Write(content); err != nil { w.Close() return fmt.Errorf("write %s: %w", name, err) } if err := w.Close(); err != nil { return fmt.Errorf("write %s: %w", name, err) } return nil } func (s *Store) Refresh() error { return s.refresh() } func (s *Store) Has(key string) bool { s.mu.RLock() defer s.mu.RUnlock() _, ok := s.entries[key] return ok } func (s *Store) Upload(folder, base, ext, mimeType string, content []byte) (string, error) { ctx := context.Background() key := folder + "/" + base name := key + "." + ext s.mu.RLock() existing, exists := s.entries[key] s.mu.RUnlock() if err := s.write(ctx, name, mimeType, content); err != nil { return "", err } if strings.HasPrefix(mimeType, "image/") { thumb, err := Thumbnail(content) if err != nil { return "", fmt.Errorf("thumbnail %s: %w", name, err) } if err := s.write(ctx, key+thumbSuffix+thumbExt, thumbMime, thumb); err != nil { return "", err } } superseded := "" if exists { superseded = fmt.Sprint(existing.generation) if existing.name != name { if err := s.bucket.Object(existing.name).Delete(ctx); err != nil { return "", fmt.Errorf("delete superseded %s: %w", existing.name, err) } } } if err := s.refresh(); err != nil { return "", fmt.Errorf("refresh after upload: %w", err) } return superseded, nil } func (s *Store) serve(w http.ResponseWriter, r *http.Request) { key := r.PathValue("folder") + "/" + r.PathValue("name") s.mu.RLock() e, ok := s.entries[key] s.mu.RUnlock() if !ok { http.NotFound(w, r) return } if r.URL.Query().Get("thumb") == "1" { if e.thumb == nil { http.NotFound(w, r) return } w.Header().Set("Content-Type", thumbMime) http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(e.thumb)) return } w.Header().Set("Content-Type", e.mimeType) http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(e.data)) } func Thumbnail(src []byte) ([]byte, error) { img, _, err := image.Decode(bytes.NewReader(src)) if err != nil { return nil, err } o := orientation(src) bounds := img.Bounds() displayWidth := bounds.Dx() if o >= 5 { displayWidth = bounds.Dy() } if displayWidth > thumbWidth { w := bounds.Dx() * thumbWidth / displayWidth h := bounds.Dy() * thumbWidth / displayWidth scaled := image.NewRGBA(image.Rect(0, 0, w, h)) draw.CatmullRom.Scale(scaled, scaled.Bounds(), img, bounds, draw.Over, nil) img = scaled } img = reorient(img, o) var buf bytes.Buffer if err := jpeg.Encode(&buf, img, &jpeg.Options{Quality: 80}); err != nil { return nil, err } return buf.Bytes(), nil } func orientation(src []byte) (o int) { o = 1 defer func() { recover() }() parsed, err := exif.Decode(bytes.NewReader(src)) if err != nil { return } tag, err := parsed.Get(exif.Orientation) if err != nil { return } value, err := tag.Int(0) if err != nil || value < 1 || value > 8 { return } return value } func reorient(img image.Image, o int) image.Image { if o == 1 { return img } b := img.Bounds() w, h := b.Dx(), b.Dy() dw, dh := w, h if o >= 5 { dw, dh = h, w } out := image.NewRGBA(image.Rect(0, 0, dw, dh)) for y := range h { for x := range w { var dx, dy int switch o { case 2: dx, dy = w-1-x, y case 3: dx, dy = w-1-x, h-1-y case 4: dx, dy = x, h-1-y case 5: dx, dy = y, x case 6: dx, dy = h-1-y, x case 7: dx, dy = h-1-y, w-1-x case 8: dx, dy = y, w-1-x } out.Set(dx, dy, img.At(b.Min.X+x, b.Min.Y+y)) } } return out }