Diffstat (limited to 'internal/storage/s3.go')
| -rw-r--r-- | internal/storage/s3.go | 377 |
1 files changed, 106 insertions, 271 deletions
diff --git a/internal/storage/s3.go b/internal/storage/s3.go index 5f3acc7..5e9bcd2 100644 --- a/internal/storage/s3.go +++ b/internal/storage/s3.go @@ -3,40 +3,31 @@ package storage import ( "bytes" "context" - "encoding/json" - "errors" "fmt" "io" - "log/slog" - "maps" - "net/url" - "slices" "strconv" "strings" - "sync" "time" "github.com/aws/aws-sdk-go-v2/aws" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/s3" - s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/SayaAndy/saya-today-article-metadata-add/config" "github.com/SayaAndy/saya-today-article-metadata-add/internal/frontmatter" ) -const s3IndexConcurrency = 32 - var _ StorageClient = &S3StorageClient{} type S3StorageClient struct { - prefix string - bucket string - client *s3.Client + prefix string + bucket string + client *s3.Client + draftModeCfg *config.DraftModeConfig } -func NewS3StorageClient(cfg *config.StorageConfig) (StorageClient, error) { +func NewS3StorageClient(cfg *config.StorageConfig, draftModeCfg *config.DraftModeConfig) (StorageClient, error) { if cfg.Type != "s3" { return nil, fmt.Errorf("invalid storage type for S3StorageClient") } @@ -68,322 +59,166 @@ func NewS3StorageClient(cfg *config.StorageConfig) (StorageClient, error) { client := s3.NewFromConfig(awsCfg, s3Opts...) + draftModeCfgCopy := *draftModeCfg + return &S3StorageClient{ - client: client, - bucket: s3cfg.BucketName, - prefix: s3cfg.Prefix, + client: client, + bucket: s3cfg.BucketName, + prefix: s3cfg.Prefix, + draftModeCfg: &draftModeCfgCopy, }, nil } -func (sc *S3StorageClient) GetMetadata(key string) (map[string]string, error) { - head, err := sc.client.HeadObject(context.Background(), &s3.HeadObjectInput{ - Bucket: aws.String(sc.bucket), - Key: aws.String(sc.prefix + key), - }) - if err != nil { - var nsk *s3types.NoSuchKey - var nf *s3types.NotFound - if errors.As(err, &nsk) || errors.As(err, &nf) { - return map[string]string{}, nil - } - return nil, fmt.Errorf("head S3 object %q: %w", sc.prefix+key, err) - } - return head.Metadata, nil -} - -func (sc *S3StorageClient) Put(key string, content []byte, metadata *frontmatter.Metadata) error { - geolocationParts := strings.Split(metadata.Geolocation, " ") - if (len(geolocationParts) == 1 && geolocationParts[0] != "") || len(geolocationParts) >= 4 { - return fmt.Errorf("invalid geolocation format, expecting '{x} {y} [areaError]' or an empty string") - } - if len(geolocationParts) >= 2 { - if _, err := strconv.ParseFloat(geolocationParts[0], 64); err != nil { - return fmt.Errorf("invalid geolocation parameter, expected float for X: %w", err) - } - if _, err := strconv.ParseFloat(geolocationParts[1], 64); err != nil { - return fmt.Errorf("invalid geolocation parameter, expected float for Y: %w", err) - } - } - if len(geolocationParts) == 3 { - if _, err := strconv.ParseFloat(geolocationParts[2], 64); err != nil { - return fmt.Errorf("invalid geolocation parameter, expected float for area error: %w", err) - } - } - - s3Metadata := map[string]string{ - "title": url.QueryEscape(metadata.Title), - "short-description": url.QueryEscape(metadata.ShortDescription), - "action-date": metadata.ActionDate, - "published-time": metadata.PublishedTime.Format(time.RFC3339), - "thumbnail": url.QueryEscape(metadata.Thumbnail), - "tags": strings.Join(metadata.Tags, ","), - "geolocation": metadata.Geolocation, - } - - targetKey := sc.prefix + key - _, err := sc.client.PutObject(context.Background(), &s3.PutObjectInput{ - Bucket: aws.String(sc.bucket), - Key: aws.String(targetKey), - Body: bytes.NewReader(content), - ContentType: aws.String("text/markdown; charset=utf-8"), - Metadata: s3Metadata, - }) - if err != nil { - return fmt.Errorf("put S3 object %q: %w", targetKey, err) - } - - return nil -} - -func (sc *S3StorageClient) BuildIndex() error { - type candidate struct { - key string - codename string - lastModified time.Time - } - var candidates []candidate +func (sc *S3StorageClient) Scan() ([]string, error) { + var filePaths []string paginator := s3.NewListObjectsV2Paginator(sc.client, &s3.ListObjectsV2Input{ Bucket: aws.String(sc.bucket), Prefix: aws.String(sc.prefix), }) + for paginator.HasMorePages() { page, err := paginator.NextPage(context.Background()) if err != nil { - return fmt.Errorf("list S3 objects: %w", err) + return nil, fmt.Errorf("list S3 objects: %w", err) } + for _, obj := range page.Contents { - key := aws.ToString(obj.Key) - if key == IndexFileName { + name := aws.ToString(obj.Key) + if !strings.HasSuffix(name, ".md") { continue } - if !strings.HasSuffix(key, ".md") { + + if sc.draftModeCfg.Enabled && !strings.HasSuffix(name, sc.draftModeCfg.DraftSuffix) { continue } - codename := key[strings.LastIndex(key, "/")+1 : strings.LastIndex(key, ".")] - candidates = append(candidates, candidate{key, codename, aws.ToTime(obj.LastModified)}) - } - } - medleys, err := sc.scanMedleys() - if err != nil { - slog.Warn("skipped reading medleys due to an error", slog.String("error", err.Error())) - } - pageToMedleyMap := make(map[string]MedleyPageEntry) - for _, medley := range medleys { - for i, page := range medley.Content { - pageToMedleyMap[page] = MedleyPageEntry{medley.Codename, i} + filePaths = append(filePaths, strings.TrimPrefix(name, sc.prefix)) } } - type result struct { - catKey string - codename string - entry IndexEntry - } - results := make([]*result, len(candidates)) - sem := make(chan struct{}, s3IndexConcurrency) - var wg sync.WaitGroup - var firstErr error - var errMu sync.Mutex - - for i, cand := range candidates { - sem <- struct{}{} - wg.Add(1) - go func() { - defer wg.Done() - defer func() { <-sem }() - - head, err := sc.client.HeadObject(context.Background(), &s3.HeadObjectInput{ - Bucket: aws.String(sc.bucket), - Key: aws.String(cand.key), - }) - if err != nil { - errMu.Lock() - if firstErr == nil { - firstErr = fmt.Errorf("head S3 object %s: %w", cand.key, err) - } - errMu.Unlock() - return - } - - if head.ContentType == nil || !strings.Contains(*head.ContentType, "text/markdown") { - return - } - meta := head.Metadata - if meta["title"] == "" { - return - } - - publishedTime, err := time.Parse(time.RFC3339, meta["published-time"]) - if err != nil { - slog.Warn("skip entry with bad published-time", slog.String("key", cand.key), slog.String("error", err.Error())) - return - } + return filePaths, nil +} - catKey := cand.key[:strings.Index(cand.key, "/")] +func (sc *S3StorageClient) GetReader(path string) (io.ReadCloser, int64, error) { + key := sc.prefix + path - tags := strings.Split(meta["tags"], ",") - slices.Sort(tags) + out, err := sc.client.GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String(sc.bucket), + Key: aws.String(key), + }) + if err != nil { + return nil, 0, fmt.Errorf("get S3 object %q: %w", key, err) + } - title, _ := url.QueryUnescape(meta["title"]) - shortDescription, _ := url.QueryUnescape(meta["short-description"]) - thumbnail, _ := url.QueryUnescape(meta["thumbnail"]) + return out.Body, aws.ToInt64(out.ContentLength), nil +} - medleyName, medleyPart := "", 0 - if medley, ok := pageToMedleyMap[cand.codename]; ok { - medleyName, medleyPart = medley.Codename, medley.Position - } +func (sc *S3StorageClient) WriteMetadata(path string, metadata *frontmatter.Metadata) error { + draftKey := sc.prefix + path - results[i] = &result{ - catKey: catKey, - codename: cand.codename, - entry: IndexEntry{ - Link: cand.key, - ModifiedTime: cand.lastModified, - Title: title, - ShortDescription: shortDescription, - ActionDate: meta["action-date"], - PublishedTime: publishedTime, - Thumbnail: thumbnail, - Tags: tags, - Geolocation: meta["geolocation"], - Medley: medleyName, - MedleyPart: medleyPart, - }, - } - }() + getOut, err := sc.client.GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String(sc.bucket), + Key: aws.String(draftKey), + }) + if err != nil { + return fmt.Errorf("get draft object %q: %w", draftKey, err) + } + content, err := io.ReadAll(getOut.Body) + getOut.Body.Close() + if err != nil { + return fmt.Errorf("read draft object %q: %w", draftKey, err) } - wg.Wait() - if firstErr != nil { - return firstErr + headOut, err := sc.client.HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String(sc.bucket), + Key: aws.String(draftKey), + }) + if err != nil { + return fmt.Errorf("head draft object %q: %w", draftKey, err) } + draftETag := aws.ToString(headOut.ETag) - now := time.Now().UTC() - fresh := make(map[string]*IndexV2Category) - for _, r := range results { - if r == nil { - continue + geolocationParts := strings.Split(metadata.Geolocation, " ") + if (len(geolocationParts) == 1 && geolocationParts[0] != "") || len(geolocationParts) >= 4 { + return fmt.Errorf("invalid geolocation format, expecting '{x} {y} [areaError]' or an empty string") + } + if len(geolocationParts) >= 2 { + if _, err := strconv.ParseFloat(geolocationParts[0], 64); err != nil { + return fmt.Errorf("invalid geolocation parameter, expected float for X: %w", err) } - if _, ok := fresh[r.catKey]; !ok { - fresh[r.catKey] = &IndexV2Category{ - Pages: make(map[string]IndexEntry), - } + if _, err := strconv.ParseFloat(geolocationParts[1], 64); err != nil { + return fmt.Errorf("invalid geolocation parameter, expected float for Y: %w", err) + } + } + if len(geolocationParts) == 3 { + if _, err := strconv.ParseFloat(geolocationParts[2], 64); err != nil { + return fmt.Errorf("invalid geolocation parameter, expected float for area error: %w", err) } - fresh[r.catKey].Pages[r.codename] = r.entry - fresh[r.catKey].GeneratedAt = now } - merged, err := sc.loadAndMergeIndex(fresh) - if err != nil { - return err + medley := "" + if metadata.Medley != "" { + medley = fmt.Sprintf("%s %d", metadata.Medley, metadata.MedleyPart) } - idx := Index{ - SchemaVersion: IndexSchemaVersion, - GeneratedAt: now, - Categories: merged, + s3Metadata := map[string]string{ + "title": metadata.Title, + "short-description": metadata.ShortDescription, + "action-date": metadata.ActionDate, + "published-time": metadata.PublishedTime.Format(time.RFC3339), + "thumbnail": metadata.Thumbnail, + "tags": strings.Join(metadata.Tags, ","), + "geolocation": metadata.Geolocation, + "medley": medley, + "metadata-last-update-etag": draftETag, } - body, err := json.Marshal(idx) - if err != nil { - return fmt.Errorf("marshal index: %w", err) + targetKey := draftKey + if sc.draftModeCfg.Enabled { + prodPath := strings.TrimSuffix(path, sc.draftModeCfg.DraftSuffix) + sc.draftModeCfg.ProdSuffix + targetKey = sc.prefix + prodPath } _, err = sc.client.PutObject(context.Background(), &s3.PutObjectInput{ Bucket: aws.String(sc.bucket), - Key: aws.String(IndexFileName), - Body: bytes.NewReader(body), - ContentType: aws.String("application/json; charset=utf-8"), + Key: aws.String(targetKey), + Body: bytes.NewReader(content), + ContentType: aws.String("text/markdown; charset=utf-8"), + Metadata: s3Metadata, }) if err != nil { - return fmt.Errorf("put index %q: %w", IndexFileName, err) + return fmt.Errorf("put S3 object %q: %w", targetKey, err) } - totalEntries := 0 - for _, c := range merged { - totalEntries += len(c.Pages) - } - slog.Info("wrote index", - slog.String("key", IndexFileName), - slog.Int("categories", len(merged)), - slog.Int("entries", totalEntries)) return nil } -func (sc *S3StorageClient) scanMedleys() ([]MedleyEntry, error) { - out, err := sc.client.GetObject(context.Background(), &s3.GetObjectInput{ +func (sc *S3StorageClient) CompareDraftAndProd(path string) bool { + draftKey := sc.prefix + path + prodPath := strings.TrimSuffix(path, sc.draftModeCfg.DraftSuffix) + sc.draftModeCfg.ProdSuffix + prodKey := sc.prefix + prodPath + + draftHead, err := sc.client.HeadObject(context.Background(), &s3.HeadObjectInput{ Bucket: aws.String(sc.bucket), - Key: aws.String(MedleysIndexFileName), + Key: aws.String(draftKey), }) if err != nil { - return nil, fmt.Errorf("get %s: %w", MedleysIndexFileName, err) - } - defer out.Body.Close() - - raw, err := io.ReadAll(out.Body) - if err != nil { - return nil, fmt.Errorf("read %s: %w", MedleysIndexFileName, err) + return false } - var entries []MedleyEntry - if err := json.Unmarshal(raw, &entries); err != nil { - return nil, fmt.Errorf("unmarshal %s: %w", MedleysIndexFileName, err) - } - - return entries, nil -} - -func (sc *S3StorageClient) loadAndMergeIndex(fresh map[string]*IndexV2Category) (map[string]*IndexV2Category, error) { - merged := make(map[string]*IndexV2Category) - - out, err := sc.client.GetObject(context.Background(), &s3.GetObjectInput{ + prodHead, err := sc.client.HeadObject(context.Background(), &s3.HeadObjectInput{ Bucket: aws.String(sc.bucket), - Key: aws.String(IndexFileName), + Key: aws.String(prodKey), }) - if err == nil { - defer out.Body.Close() - raw, readErr := io.ReadAll(out.Body) - if readErr != nil { - return nil, fmt.Errorf("read existing index: %w", readErr) - } - - var legacyIdx Index - if err := json.Unmarshal(raw, &legacyIdx); err != nil { - return nil, fmt.Errorf("unmarshal existing index: %w", err) - } - - switch legacyIdx.SchemaVersion { - case 1: - for k, v := range *legacyIdx.Categories.(*map[string]*IndexV1Category) { - pages := make(map[string]IndexEntry, len(v.Pages)) - for _, page := range v.Pages { - pages[page.Link[strings.LastIndex(page.Link, "/")+1:strings.LastIndex(page.Link, ".")]] = page - } - merged[k] = &IndexV2Category{ - GeneratedAt: v.GeneratedAt, - Pages: pages, - } - } - case 2: - maps.Copy(merged, *legacyIdx.Categories.(*map[string]*IndexV2Category)) - default: - slog.Warn("unknown index schema, discarding", slog.Int("schema_version", legacyIdx.SchemaVersion)) - } - } else { - var nsk *s3types.NoSuchKey - if !errors.As(err, &nsk) { - return nil, fmt.Errorf("get existing index: %w", err) - } + if err != nil { + return true } - for k := range merged { - if strings.HasPrefix(k, sc.prefix) { - delete(merged, k) - } + lastUpdateETag, ok := prodHead.Metadata["metadata-last-update-etag"] + if !ok { + return true } - maps.Copy(merged, fresh) - return merged, nil + return aws.ToString(draftHead.ETag) != lastUpdateETag } |