feat: repurpose metadata parser as article transcoder for saya.uz andmain

telegram

1 files changed, 271 insertions, 106 deletions

diff --git a/internal/storage/s3.go b/internal/storage/s3.go
index 5e9bcd2..5f3acc7 100644
--- a/internal/storage/s3.go
+++ b/internal/storage/s3.go
@@ -3,31 +3,40 @@ 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
-	draftModeCfg *config.DraftModeConfig
+	prefix string
+	bucket string
+	client *s3.Client
 }
 
-func NewS3StorageClient(cfg *config.StorageConfig, draftModeCfg *config.DraftModeConfig) (StorageClient, error) {
+func NewS3StorageClient(cfg *config.StorageConfig) (StorageClient, error) {
 	if cfg.Type != "s3" {
 		return nil, fmt.Errorf("invalid storage type for S3StorageClient")
 	}
@@ -59,166 +68,322 @@ func NewS3StorageClient(cfg *config.StorageConfig, draftModeCfg *config.DraftMod
 
 	client := s3.NewFromConfig(awsCfg, s3Opts...)
 
-	draftModeCfgCopy := *draftModeCfg
-
 	return &S3StorageClient{
-		client:       client,
-		bucket:       s3cfg.BucketName,
-		prefix:       s3cfg.Prefix,
-		draftModeCfg: &draftModeCfgCopy,
+		client: client,
+		bucket: s3cfg.BucketName,
+		prefix: s3cfg.Prefix,
 	}, nil
 }
 
-func (sc *S3StorageClient) Scan() ([]string, error) {
-	var filePaths []string
+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 nil, fmt.Errorf("list S3 objects: %w", err)
+			return fmt.Errorf("list S3 objects: %w", err)
 		}
-
 		for _, obj := range page.Contents {
-			name := aws.ToString(obj.Key)
-			if !strings.HasSuffix(name, ".md") {
+			key := aws.ToString(obj.Key)
+			if key == IndexFileName {
 				continue
 			}
-
-			if sc.draftModeCfg.Enabled && !strings.HasSuffix(name, sc.draftModeCfg.DraftSuffix) {
+			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)})
+		}
+	}
 
-			filePaths = append(filePaths, strings.TrimPrefix(name, sc.prefix))
+	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}
 		}
 	}
 
-	return filePaths, nil
-}
+	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
 
-func (sc *S3StorageClient) GetReader(path string) (io.ReadCloser, int64, error) {
-	key := sc.prefix + path
+	for i, cand := range candidates {
+		sem <- struct{}{}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			defer func() { <-sem }()
 
-	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)
-	}
+			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
+			}
 
-	return out.Body, aws.ToInt64(out.ContentLength), nil
-}
+			if head.ContentType == nil || !strings.Contains(*head.ContentType, "text/markdown") {
+				return
+			}
+			meta := head.Metadata
+			if meta["title"] == "" {
+				return
+			}
 
-func (sc *S3StorageClient) WriteMetadata(path string, metadata *frontmatter.Metadata) error {
-	draftKey := sc.prefix + path
+			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
+			}
 
-	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)
-	}
+			catKey := cand.key[:strings.Index(cand.key, "/")]
 
-	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)
+			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,
+				},
+			}
+		}()
 	}
-	draftETag := aws.ToString(headOut.ETag)
+	wg.Wait()
 
-	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 firstErr != nil {
+		return firstErr
 	}
-	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)
+
+	now := time.Now().UTC()
+	fresh := make(map[string]*IndexV2Category)
+	for _, r := range results {
+		if r == nil {
+			continue
 		}
-	}
-	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)
+		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
 	}
 
-	medley := ""
-	if metadata.Medley != "" {
-		medley = fmt.Sprintf("%s %d", metadata.Medley, metadata.MedleyPart)
+	merged, err := sc.loadAndMergeIndex(fresh)
+	if err != nil {
+		return err
 	}
 
-	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,
+	idx := Index{
+		SchemaVersion: IndexSchemaVersion,
+		GeneratedAt:   now,
+		Categories:    merged,
 	}
 
-	targetKey := draftKey
-	if sc.draftModeCfg.Enabled {
-		prodPath := strings.TrimSuffix(path, sc.draftModeCfg.DraftSuffix) + sc.draftModeCfg.ProdSuffix
-		targetKey = sc.prefix + prodPath
+	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(targetKey),
-		Body:        bytes.NewReader(content),
-		ContentType: aws.String("text/markdown; charset=utf-8"),
-		Metadata:    s3Metadata,
+		Key:         aws.String(IndexFileName),
+		Body:        bytes.NewReader(body),
+		ContentType: aws.String("application/json; charset=utf-8"),
 	})
 	if err != nil {
-		return fmt.Errorf("put S3 object %q: %w", targetKey, err)
+		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) 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{
+func (sc *S3StorageClient) scanMedleys() ([]MedleyEntry, error) {
+	out, err := sc.client.GetObject(context.Background(), &s3.GetObjectInput{
 		Bucket: aws.String(sc.bucket),
-		Key:    aws.String(draftKey),
+		Key:    aws.String(MedleysIndexFileName),
 	})
 	if err != nil {
-		return false
+		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)
 	}
 
-	prodHead, err := sc.client.HeadObject(context.Background(), &s3.HeadObjectInput{
+	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(prodKey),
+		Key:    aws.String(IndexFileName),
 	})
-	if err != nil {
-		return true
+	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)
+		}
 	}
 
-	lastUpdateETag, ok := prodHead.Metadata["metadata-last-update-etag"]
-	if !ok {
-		return true
+	for k := range merged {
+		if strings.HasPrefix(k, sc.prefix) {
+			delete(merged, k)
+		}
 	}
+	maps.Copy(merged, fresh)
 
-	return aws.ToString(draftHead.ETag) != lastUpdateETag
+	return merged, nil
 }