summaryrefslogtreecommitdiff
path: root/internal/storage
diff refs
from: back
to: back
| flip
diff options
context:
space:
mode:
Diffstat (limited to 'internal/storage')
-rw-r--r--internal/storage/b2.go134
-rw-r--r--internal/storage/index.go92
-rw-r--r--internal/storage/s3.go389
-rw-r--r--internal/storage/storage_interface.go12
4 files changed, 538 insertions, 89 deletions
diff --git a/internal/storage/b2.go b/internal/storage/b2.go
index 168690a..6697023 100644
--- a/internal/storage/b2.go
+++ b/internal/storage/b2.go
@@ -3,7 +3,7 @@ package storage
import (
"context"
"fmt"
- "io"
+ "strconv"
"strings"
"time"
@@ -39,102 +39,72 @@ func NewB2StorageClient(cfg *config.StorageConfig) (StorageClient, error) {
return &B2StorageClient{b2cl: b2cl, bucket: bucket, prefix: b2cfg.Prefix}, nil
}
-func (sc *B2StorageClient) Scan() ([]string, error) {
- filePaths := []string{}
-
- iter := sc.bucket.List(context.Background(), b2.ListPrefix(sc.prefix))
-
- for iter.Next() {
- obj := iter.Object()
- if obj == nil {
- return nil, fmt.Errorf("failed to reference object in B2 bucket")
- }
-
- attrs, err := obj.Attrs(context.Background())
- if err != nil {
- return nil, fmt.Errorf("get attributes for object: %w", err)
- }
-
- if attrs.Status != b2.Uploaded {
- continue
- }
-
- name := obj.Name()
-
- nameParts := strings.Split(name, ".")
- if len(nameParts) < 2 {
- continue
- }
-
- ext := strings.ToLower(nameParts[len(nameParts)-1])
- if ext != "md" {
- continue
- }
-
- filePaths = append(filePaths, strings.TrimPrefix(name, sc.prefix))
- }
-
- if err := iter.Err(); err != nil {
- return nil, fmt.Errorf("iterate over B2 objects: %w", err)
- }
-
- return filePaths, nil
-}
-
-func (sc *B2StorageClient) GetReader(path string) (io.ReadCloser, int64, error) {
- obj := sc.bucket.Object(sc.prefix + path)
+func (sc *B2StorageClient) GetMetadata(key string) (map[string]string, error) {
+ obj := sc.bucket.Object(sc.prefix + key)
if obj == nil {
- return nil, 0, fmt.Errorf("failed to reference object in B2 bucket")
+ return nil, fmt.Errorf("failed to reference object in B2 bucket")
}
attrs, err := obj.Attrs(context.Background())
if err != nil {
- return nil, 0, fmt.Errorf("error getting attributes of an object: %w", err)
+ if b2.IsNotExist(err) {
+ return map[string]string{}, nil
+ }
+ return nil, fmt.Errorf("get attributes of B2 object %q: %w", sc.prefix+key, err)
}
-
- return obj.NewReader(context.Background()), attrs.Size, nil
+ return attrs.Info, nil
}
-func (sc *B2StorageClient) WriteMetadata(path string, metadata *frontmatter.Metadata) error {
- obj := sc.bucket.Object(sc.prefix + path)
- if obj == nil {
- return fmt.Errorf("failed to reference object in B2 bucket")
+func (sc *B2StorageClient) 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")
}
- oldAttrs, err := obj.Attrs(context.Background())
- if err != nil {
- return fmt.Errorf("error getting attributes of an object: %w", err)
+ 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)
+ }
}
- attrs := &b2.Attrs{Info: 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, ","),
- "metadata-last-update-sha1": oldAttrs.SHA1,
- }}
-
- writer := obj.NewWriter(context.Background(), b2.WithAttrsOption(attrs))
- writer.Close()
+ medley := ""
+ if metadata.Medley != "" {
+ medley = fmt.Sprintf("%s %d", metadata.Medley, metadata.MedleyPart)
+ }
- return nil
-}
+ attrs := &b2.Attrs{
+ ContentType: "text/markdown; charset=utf-8",
+ Info: 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,
+ }}
-func (sc *B2StorageClient) FileHasChanged(path string) bool {
- obj := sc.bucket.Object(sc.prefix + path)
- if obj == nil {
- return true
+ prod := sc.bucket.Object(sc.prefix + key)
+ if prod == nil {
+ return fmt.Errorf("failed to reference prod object in B2 bucket")
}
- attrs, err := obj.Attrs(context.Background())
- if err != nil {
- return true
+ writer := prod.NewWriter(context.Background(), b2.WithAttrsOption(attrs))
+ defer writer.Close()
+ if _, err := writer.Write(content); err != nil {
+ return fmt.Errorf("failed to write an object back after attribute settings: %w", err)
}
- lastUpdateSha1, ok := attrs.Info["metadata-last-update-sha1"]
- if !ok {
- return true
- }
+ return nil
+}
- return attrs.SHA1 == lastUpdateSha1
+func (sc *B2StorageClient) BuildIndex() error {
+ return fmt.Errorf("BuildIndex not implemented for B2 storage")
}
diff --git a/internal/storage/index.go b/internal/storage/index.go
new file mode 100644
index 0000000..bfeafbe
--- /dev/null
+++ b/internal/storage/index.go
@@ -0,0 +1,92 @@
+package storage
+
+import (
+ "encoding/json"
+ "fmt"
+ "time"
+)
+
+const IndexFileName = "index.json"
+const MedleysIndexFileName = "medleys.json"
+
+const IndexSchemaVersion = 2
+
+type IndexEntry struct {
+ Link string `json:"link"`
+ ModifiedTime time.Time `json:"modifiedTime"`
+ Title string `json:"title"`
+ ShortDescription string `json:"shortDescription"`
+ ActionDate string `json:"actionDate"`
+ PublishedTime time.Time `json:"publishedTime"`
+ Thumbnail string `json:"thumbnail"`
+ Tags []string `json:"tags"`
+ Geolocation string `json:"geolocation"`
+ Medley string `json:"medley,omitempty"`
+ MedleyPart int `json:"medleyPart,omitempty"`
+}
+
+type IndexV2Category struct {
+ GeneratedAt time.Time `json:"generatedAt"`
+ Pages map[string]IndexEntry `json:"pages"`
+}
+
+type IndexV1Category struct {
+ GeneratedAt time.Time `json:"generatedAt"`
+ Pages []IndexEntry `json:"pages"`
+}
+
+type Index struct {
+ SchemaVersion int `json:"schemaVersion"`
+ GeneratedAt time.Time `json:"generatedAt"`
+ Categories any `json:"categories"`
+}
+
+func (idx *Index) UnmarshalJSON(data []byte) error {
+ var tmp struct {
+ SchemaVersion int `json:"schemaVersion"`
+ GeneratedAt time.Time `json:"generatedAt"`
+ Categories json.RawMessage `json:"categories"`
+ }
+
+ if err := json.Unmarshal(data, &tmp); err != nil {
+ return err
+ }
+
+ idx.SchemaVersion = tmp.SchemaVersion
+ idx.GeneratedAt = tmp.GeneratedAt
+
+ switch tmp.SchemaVersion {
+ case 1:
+ var categories map[string]*IndexV1Category
+ if err := json.Unmarshal(tmp.Categories, &categories); err != nil {
+ return fmt.Errorf("unmarshal map[string]*IndexV1Category: %w", err)
+ }
+ idx.Categories = &categories
+ case 2:
+ var categories map[string]*IndexV2Category
+ if err := json.Unmarshal(tmp.Categories, &categories); err != nil {
+ return fmt.Errorf("unmarshal map[string]*IndexV2Category: %w", err)
+ }
+ idx.Categories = &categories
+ default:
+ return fmt.Errorf("unsupported index version: %d", tmp.SchemaVersion)
+ }
+
+ return nil
+}
+
+type IndexV1 struct {
+ SchemaVersion int `json:"schemaVersion"`
+ GeneratedAt time.Time `json:"generatedAt"`
+ Categories map[string]*IndexV1Category `json:"categories"`
+}
+
+type MedleyEntry struct {
+ Codename string `json:"codename"`
+ Content []string `json:"content"`
+}
+
+type MedleyPageEntry struct {
+ Codename string `json:"codename"`
+ Position int `json:"position"`
+}
diff --git a/internal/storage/s3.go b/internal/storage/s3.go
new file mode 100644
index 0000000..5f3acc7
--- /dev/null
+++ b/internal/storage/s3.go
@@ -0,0 +1,389 @@
+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
+}
+
+func NewS3StorageClient(cfg *config.StorageConfig) (StorageClient, error) {
+ if cfg.Type != "s3" {
+ return nil, fmt.Errorf("invalid storage type for S3StorageClient")
+ }
+ s3cfg := cfg.Config.(*config.S3Config)
+
+ opts := []func(*awsconfig.LoadOptions) error{
+ awsconfig.WithRegion(s3cfg.Region),
+ }
+ if s3cfg.AccessKeyID != "" && s3cfg.SecretAccessKey != "" {
+ opts = append(opts, awsconfig.WithCredentialsProvider(
+ credentials.NewStaticCredentialsProvider(s3cfg.AccessKeyID, s3cfg.SecretAccessKey, ""),
+ ))
+ }
+
+ awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...)
+ if err != nil {
+ return nil, fmt.Errorf("load AWS config: %w", err)
+ }
+
+ var s3Opts []func(*s3.Options)
+ if s3cfg.Endpoint != "" {
+ s3Opts = append(s3Opts, func(o *s3.Options) {
+ o.BaseEndpoint = aws.String(s3cfg.Endpoint)
+ })
+ }
+ s3Opts = append(s3Opts, func(o *s3.Options) {
+ o.UsePathStyle = s3cfg.UsePathStyle
+ })
+
+ client := s3.NewFromConfig(awsCfg, s3Opts...)
+
+ return &S3StorageClient{
+ client: client,
+ bucket: s3cfg.BucketName,
+ prefix: s3cfg.Prefix,
+ }, 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
+
+ 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)
+ }
+ for _, obj := range page.Contents {
+ key := aws.ToString(obj.Key)
+ if key == IndexFileName {
+ continue
+ }
+ if !strings.HasSuffix(key, ".md") {
+ 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}
+ }
+ }
+
+ 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
+ }
+
+ catKey := cand.key[:strings.Index(cand.key, "/")]
+
+ tags := strings.Split(meta["tags"], ",")
+ slices.Sort(tags)
+
+ title, _ := url.QueryUnescape(meta["title"])
+ shortDescription, _ := url.QueryUnescape(meta["short-description"])
+ thumbnail, _ := url.QueryUnescape(meta["thumbnail"])
+
+ medleyName, medleyPart := "", 0
+ if medley, ok := pageToMedleyMap[cand.codename]; ok {
+ medleyName, medleyPart = medley.Codename, medley.Position
+ }
+
+ 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,
+ },
+ }
+ }()
+ }
+ wg.Wait()
+
+ if firstErr != nil {
+ return firstErr
+ }
+
+ now := time.Now().UTC()
+ fresh := make(map[string]*IndexV2Category)
+ for _, r := range results {
+ if r == nil {
+ continue
+ }
+ if _, ok := fresh[r.catKey]; !ok {
+ fresh[r.catKey] = &IndexV2Category{
+ Pages: make(map[string]IndexEntry),
+ }
+ }
+ fresh[r.catKey].Pages[r.codename] = r.entry
+ fresh[r.catKey].GeneratedAt = now
+ }
+
+ merged, err := sc.loadAndMergeIndex(fresh)
+ if err != nil {
+ return err
+ }
+
+ idx := Index{
+ SchemaVersion: IndexSchemaVersion,
+ GeneratedAt: now,
+ Categories: merged,
+ }
+
+ body, err := json.Marshal(idx)
+ if err != nil {
+ return fmt.Errorf("marshal index: %w", err)
+ }
+
+ _, 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"),
+ })
+ if err != nil {
+ return fmt.Errorf("put index %q: %w", IndexFileName, 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{
+ Bucket: aws.String(sc.bucket),
+ Key: aws.String(MedleysIndexFileName),
+ })
+ 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)
+ }
+
+ 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{
+ Bucket: aws.String(sc.bucket),
+ Key: aws.String(IndexFileName),
+ })
+ 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)
+ }
+ }
+
+ for k := range merged {
+ if strings.HasPrefix(k, sc.prefix) {
+ delete(merged, k)
+ }
+ }
+ maps.Copy(merged, fresh)
+
+ return merged, nil
+}
diff --git a/internal/storage/storage_interface.go b/internal/storage/storage_interface.go
index 91eeedb..92a37e2 100644
--- a/internal/storage/storage_interface.go
+++ b/internal/storage/storage_interface.go
@@ -1,19 +1,17 @@
package storage
import (
- "io"
-
"github.com/SayaAndy/saya-today-article-metadata-add/config"
"github.com/SayaAndy/saya-today-article-metadata-add/internal/frontmatter"
)
type StorageClient interface {
- Scan() (paths []string, err error)
- GetReader(path string) (reader io.ReadCloser, sz int64, err error)
- WriteMetadata(path string, metadata *frontmatter.Metadata) error
- FileHasChanged(path string) bool
+ Put(key string, content []byte, metadata *frontmatter.Metadata) error
+ GetMetadata(key string) (metadata map[string]string, err error)
+ BuildIndex() error
}
-var NewStorageClientMap = map[string]func(cfg *config.StorageConfig) (StorageClient, error){
+var NewStorageClientMap = map[string]func(*config.StorageConfig) (StorageClient, error){
"b2": NewB2StorageClient,
+ "s3": NewS3StorageClient,
}
generated by cgitext v1.3.1+17 (git 2.54.0) at 2026-09-17 22:12:34 +0000