query-orchestration/internal/document/store/process.go

package documentstore

import (
	"context"
	"log/slog"
	"net/url"

	docinitrunner "queryorchestration/api/docInitRunner"
	"queryorchestration/internal/serviceconfig/objectstore"
	"queryorchestration/internal/serviceconfig/queue"
)

type Params struct {
	Event  EventS3
	Key    string
	Bucket string
	Hash   string
}

type EventS3 string

const (
	EventS3ObjectCreatedPut                     = "ObjectCreated:Put"
	EventS3ObjectCreatedCompleteMultipartUpload = "ObjectCreated:CompleteMultipartUpload"
)

func (s *Service) Process(ctx context.Context, params Params) error {
	if !s.isSupportedEvent(params.Event) {
		slog.Warn("unsupported event", "name", params.Event)
		return nil
	}

	// URL decode the key (AWS S3 sends URL-encoded keys, LocalStack doesn't)
	decodedKey, err := url.QueryUnescape(params.Key)
	if err != nil {
		slog.Error("unable to decode key", "key", params.Key, "error", err)
		return nil
	}

	key, err := objectstore.ParseBucketKey(decodedKey)
	if err != nil {
		slog.Error("unable to parse key", "key", decodedKey, "original_key", params.Key, "error", err)
		return nil
	}

	queueParams := &queue.SendParams{}
	switch key.Location {
	case objectstore.Import:
		queueParams.QueueURL = s.cfg.GetDocInitURL()
		queueParams.Body = docinitrunner.Body{
			Bucket: params.Bucket,
			Key:    decodedKey,
			Hash:   params.Hash,
		}
	default:
		slog.Info("unsupported key", "key", decodedKey)
		return nil
	}

	return s.cfg.SendToQueue(ctx, queueParams)
}

func (s *Service) isSupportedEvent(name EventS3) bool {
	return name == EventS3ObjectCreatedPut ||
		name == EventS3ObjectCreatedCompleteMultipartUpload
}