tm_back/pkg/ai_summarizer/storage.go

package ai_summarizer

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"path/filepath"
	"strings"
	"time"

	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"

	"tm/pkg/logger"
)

// procedurePrefix is prepended to every contract folder id in the MinIO key.
// The AI team's storage layout in bucket opplens-documents is:
//
//	PROC_<contractFolderID>/<noticePublicationID>/tender.json
//	PROC_<contractFolderID>/<noticePublicationID>/documents/{files}
const procedurePrefix = "PROC_"

// StorageClient wraps a MinIO client configured to connect to the AI service's
// MinIO bucket and retrieve tender.json files.
type StorageClient struct {
	client *minio.Client
	config *Config
	logger logger.Logger
}

// TenderSummaryItem represents a single tender's summary retrieved from storage.
type TenderSummaryItem struct {
	ContractFolderID string `json:"contract_folder_id"`
	NoticeID         string `json:"notice_id"`
	OverallSummary   string `json:"overall_summary"`
	Done             bool   `json:"done"`
}

// noticeKeyVariants returns MinIO notice folder names to try. The pipeline may
// store either the raw TED id (e.g. "00175735-2026") or a zero-stripped form
// ("175735-2026"); we try the raw value first, then the normalized variant.
func noticeKeyVariants(noticePublicationID string) []string {
	raw := strings.TrimSpace(noticePublicationID)
	if raw == "" {
		return nil
	}
	normalized := normalizeNoticeKey(raw)
	if normalized == raw {
		return []string{raw}
	}
	return []string{raw, normalized}
}

// noticeObjectPrefix returns the MinIO key prefix for one notice folder name.
func noticeObjectPrefix(contractFolderID, noticeKey string) string {
	return fmt.Sprintf(
		"%s%s/%s/",
		procedurePrefix,
		normalizeContractFolderID(contractFolderID),
		strings.TrimSpace(noticeKey),
	)
}

// noticeObjectPrefixes returns every prefix variant for a notice publication id.
func noticeObjectPrefixes(contractFolderID, noticePublicationID string) []string {
	folderID := normalizeContractFolderID(contractFolderID)
	keys := noticeKeyVariants(noticePublicationID)
	prefixes := make([]string, 0, len(keys))
	seen := make(map[string]struct{}, len(keys))
	for _, key := range keys {
		prefix := noticeObjectPrefix(folderID, key)
		if _, dup := seen[prefix]; dup {
			continue
		}
		seen[prefix] = struct{}{}
		prefixes = append(prefixes, prefix)
	}
	return prefixes
}

// tenderJSONKeyVariants returns object keys to try for tender.json.
func tenderJSONKeyVariants(contractFolderID, noticePublicationID string) []string {
	prefixes := noticeObjectPrefixes(contractFolderID, noticePublicationID)
	keys := make([]string, 0, len(prefixes))
	for _, prefix := range prefixes {
		keys = append(keys, prefix+"tender.json")
	}
	return keys
}

// normalizeContractFolderID strips an accidental PROC_ prefix from stored ids.
func normalizeContractFolderID(contractFolderID string) string {
	id := strings.TrimSpace(contractFolderID)
	if len(id) >= 5 && strings.EqualFold(id[:5], "proc_") {
		return strings.TrimSpace(id[5:])
	}
	return id
}

// normalizeNoticeKey trims the leading zeros that we add when persisting the
// notice publication id (e.g. "00322263-2026") so it matches the AI team's
// folder convention ("322263-2026"). Empty input is preserved.
func normalizeNoticeKey(noticePublicationID string) string {
	id := strings.TrimSpace(noticePublicationID)
	if id == "" {
		return ""
	}
	// Strip leading zeros from the numeric portion only; keep the year suffix
	// (and any non-numeric tail) intact.
	out := strings.TrimLeft(id, "0")
	if out == "" || strings.HasPrefix(out, "-") {
		// All-zero numeric portion: keep a single "0" so the key is never empty
		// and never starts with "-".
		return "0" + out
	}
	return out
}

// NewStorageClient creates a new MinIO storage client for the AI service.
// It establishes a connection to the external MinIO instance and verifies
// that the configured bucket exists.
func NewStorageClient(config *Config, log logger.Logger) (*StorageClient, error) {
	if config == nil {
		config = DefaultConfig()
	}
	if err := config.ValidateStorage(); err != nil {
		return nil, fmt.Errorf("ai_summarizer: invalid storage config: %w", err)
	}

	log.Info("Connecting to AI summarizer MinIO", map[string]interface{}{
		"endpoint": config.MinioEndpoint,
		"use_ssl":  config.MinioUseSSL,
		"region":   config.MinioRegion,
		"bucket":   config.MinioBucket,
	})

	minioClient, err := minio.New(config.MinioEndpoint, &minio.Options{
		Creds:  credentials.NewStaticV4(config.MinioAccessKey, config.MinioSecretKey, ""),
		Secure: config.MinioUseSSL,
		Region: config.MinioRegion,
	})
	if err != nil {
		log.Error("Failed to create AI summarizer MinIO client", map[string]interface{}{
			"endpoint": config.MinioEndpoint,
			"error":    err.Error(),
		})
		return nil, fmt.Errorf("ai_summarizer: failed to create MinIO client: %w", err)
	}

	storage := &StorageClient{
		client: minioClient,
		config: config,
		logger: log,
	}

	timeout := config.MinioTimeout
	if timeout <= 0 {
		timeout = 30 * time.Second
	}
	verifyCtx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	if err := storage.verifyConnection(verifyCtx); err != nil {
		log.Error("AI summarizer MinIO connection check failed", map[string]interface{}{
			"endpoint": config.MinioEndpoint,
			"bucket":   config.MinioBucket,
			"error":    err.Error(),
		})
		return nil, fmt.Errorf("ai_summarizer: MinIO connection check failed: %w", err)
	}

	log.Info("AI summarizer MinIO connection verified", map[string]interface{}{
		"endpoint": config.MinioEndpoint,
		"bucket":   config.MinioBucket,
	})

	return storage, nil
}

// verifyConnection checks that MinIO is reachable and the configured bucket exists.
func (s *StorageClient) verifyConnection(ctx context.Context) error {
	exists, err := s.client.BucketExists(ctx, s.config.MinioBucket)
	if err != nil {
		return fmt.Errorf("bucket exists check for %q: %w", s.config.MinioBucket, err)
	}
	if !exists {
		return fmt.Errorf("%w: %q", ErrBucketNotFound, s.config.MinioBucket)
	}

	buckets, err := s.client.ListBuckets(ctx)
	if err != nil {
		s.logger.Warn("MinIO bucket exists but ListBuckets failed (credentials may be bucket-scoped)", map[string]interface{}{
			"endpoint": s.config.MinioEndpoint,
			"bucket":   s.config.MinioBucket,
			"error":    err.Error(),
		})
		return nil
	}

	names := make([]string, 0, len(buckets))
	for _, b := range buckets {
		names = append(names, b.Name)
	}
	s.logger.Debug("MinIO ListBuckets succeeded", map[string]interface{}{
		"endpoint":      s.config.MinioEndpoint,
		"bucket_count":  len(names),
		"target_bucket": s.config.MinioBucket,
	})
	return nil
}

// Ping checks that MinIO is still reachable and the configured bucket exists.
func (s *StorageClient) Ping(ctx context.Context) error {
	return s.verifyConnection(ctx)
}

// resolveNoticeStoragePrefix locates the MinIO folder for a notice. It first
// checks PROC_<contractFolderID>/<notice>/ using known id variants, then falls
// back to a bucket search by notice publication id when the stored contract
// folder id does not match the AI pipeline folder (common in local testing).
func (s *StorageClient) resolveNoticeStoragePrefix(ctx context.Context, contractFolderID, noticePublicationID string) (string, error) {
	for _, prefix := range noticeObjectPrefixes(contractFolderID, noticePublicationID) {
		_, err := s.client.StatObject(ctx, s.config.MinioBucket, prefix+"tender.json", minio.StatObjectOptions{})
		if err == nil {
			return prefix, nil
		}
		if !isMinioNotFound(err) {
			return "", fmt.Errorf("ai_summarizer: stat tender.json under %s: %w", prefix, err)
		}
	}

	for _, noticeKey := range noticeKeyVariants(noticePublicationID) {
		marker := "/" + noticeKey + "/"
		var resolved string
		objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
			Prefix:    procedurePrefix,
			Recursive: true,
		})
		for object := range objectCh {
			if object.Err != nil {
				return "", fmt.Errorf("ai_summarizer: search notice folder: %w", object.Err)
			}
			if idx := strings.Index(object.Key, marker); idx >= 0 {
				resolved = object.Key[:idx+len(marker)]
				break
			}
		}
		if resolved != "" {
			s.logger.Info("Resolved MinIO notice folder by notice publication id", map[string]interface{}{
				"notice_id":                    noticePublicationID,
				"requested_contract_folder_id": contractFolderID,
				"resolved_prefix":              resolved,
			})
			return resolved, nil
		}
	}

	s.logger.Debug("No MinIO folder found for notice", map[string]interface{}{
		"contract_folder_id": contractFolderID,
		"notice_id":          noticePublicationID,
	})
	return "", ErrObjectNotFound
}

// GetSummaryFromStorage fetches the
// PROC_<contractFolderID>/<noticePublicationID>/tender.json file from the AI
// service's MinIO bucket, parses it, and returns the overall_summary string.
//
// It returns:
//   - (summary, nil) on success
//   - ("", ErrObjectNotFound) if the tender.json does not exist (404)
//   - ("", ErrSummaryNotReady) if the file exists but summarize_status.done is false
//   - ("", ErrNoOverallSummary) if done is true but overall_summary is empty/null
//   - ("", error) for any other failure
func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) (string, error) {
	if strings.TrimSpace(contractFolderID) == "" {
		return "", fmt.Errorf("ai_summarizer: contractFolderID is required")
	}
	if strings.TrimSpace(noticePublicationID) == "" {
		return "", fmt.Errorf("ai_summarizer: noticePublicationID is required")
	}
	s.logger.Debug("Fetching tender.json from MinIO", map[string]interface{}{
		"bucket":             s.config.MinioBucket,
		"contract_folder_id": contractFolderID,
		"notice_id":          noticePublicationID,
	})

	prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
	if err != nil {
		return "", err
	}
	return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID)
}

// getSummaryFromObjectKey reads tender.json at objectKey and returns overall_summary when ready.
// contractFolderID and noticePublicationID are used for logging only (may be empty when unknown).
func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) {
	object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectKey, minio.GetObjectOptions{})
	if err != nil {
		if isMinioNotFound(err) {
			s.logger.Debug("MinIO object not found", map[string]interface{}{
				"bucket":     s.config.MinioBucket,
				"object_key": objectKey,
			})
			return "", ErrObjectNotFound
		}
		var minioErr minio.ErrorResponse
		if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusForbidden {
			return "", fmt.Errorf("ai_summarizer: access denied to bucket %s: %w", s.config.MinioBucket, err)
		}
		return "", fmt.Errorf("ai_summarizer: failed to get object: %w", err)
	}
	defer object.Close()

	data, err := io.ReadAll(object)
	if err != nil {
		if isMinioNotFound(err) {
			s.logger.Debug("MinIO object not found on read", map[string]interface{}{
				"bucket":     s.config.MinioBucket,
				"object_key": objectKey,
			})
			return "", ErrObjectNotFound
		}
		return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err)
	}

	var tenderJSON TenderJSON
	if err := json.Unmarshal(data, &tenderJSON); err != nil {
		return "", fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
	}

	done, overall := tenderJSON.resolved()
	if !done {
		return "", ErrSummaryNotReady
	}
	if overall == nil || strings.TrimSpace(*overall) == "" {
		return "", ErrNoOverallSummary
	}

	summaryText := strings.TrimSpace(*overall)

	s.logger.Info("Retrieved overall_summary from storage", map[string]interface{}{
		"contract_folder_id": contractFolderID,
		"notice_id":          noticePublicationID,
		"object_key":         objectKey,
		"summary_length":     len(summaryText),
	})

	return summaryText, nil
}

// IsSummaryReady is a convenience method that checks whether the tender.json
// exists and the summarize_status.done flag is true, without returning the
// actual summary text.
func (s *StorageClient) IsSummaryReady(ctx context.Context, contractFolderID, noticePublicationID string) (bool, error) {
	if strings.TrimSpace(contractFolderID) == "" || strings.TrimSpace(noticePublicationID) == "" {
		return false, fmt.Errorf("ai_summarizer: contractFolderID and noticePublicationID are required")
	}
	prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
	if errors.Is(err, ErrObjectNotFound) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	objectPath := prefix + "tender.json"
	object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectPath, minio.GetObjectOptions{})
	if err != nil {
		if isMinioNotFound(err) {
			return false, nil
		}
		return false, fmt.Errorf("ai_summarizer: failed to check summary readiness: %w", err)
	}

	data, readErr := io.ReadAll(object)
	closeErr := object.Close()
	if readErr != nil {
		if isMinioNotFound(readErr) {
			return false, nil
		}
		return false, fmt.Errorf("ai_summarizer: failed to read tender.json: %w", readErr)
	}
		if closeErr != nil {
			s.logger.Warn("Failed to close tender.json object after readiness check", map[string]interface{}{
				"object_key": objectPath,
				"error":      closeErr.Error(),
			})
		}

	var tenderJSON TenderJSON
	if err := json.Unmarshal(data, &tenderJSON); err != nil {
		return false, fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err)
	}

	done, _ := tenderJSON.resolved()
	return done, nil
}

// ListDocuments lists document files stored under
// PROC_<contractFolderID>/<noticePublicationID>/. Non-document artefacts
// (tender.json, AI cache files, etc.) are filtered out.
func (s *StorageClient) ListDocuments(ctx context.Context, contractFolderID, noticePublicationID string) ([]StoredDocument, error) {
	if strings.TrimSpace(contractFolderID) == "" {
		return nil, fmt.Errorf("ai_summarizer: contractFolderID is required")
	}
	if strings.TrimSpace(noticePublicationID) == "" {
		return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required")
	}

	prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
	if errors.Is(err, ErrObjectNotFound) {
		s.logger.Info("Listed tender documents from storage", map[string]interface{}{
			"contract_folder_id": contractFolderID,
			"notice_id":          noticePublicationID,
			"documents_count":    0,
		})
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	prefixes := []string{prefix}

	s.logger.Debug("Listing tender documents from MinIO", map[string]interface{}{
		"bucket":   s.config.MinioBucket,
		"prefixes": prefixes,
	})

	documents := make([]StoredDocument, 0)
	seenObjects := make(map[string]struct{})
	for _, prefix := range prefixes {
		objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
			Prefix:    prefix,
			Recursive: true,
		})

		for object := range objectCh {
			if object.Err != nil {
				return nil, fmt.Errorf("ai_summarizer: failed to list documents under %s: %w", prefix, object.Err)
			}
			if !isTenderDocumentObject(prefix, object.Key) {
				continue
			}
			if _, dup := seenObjects[object.Key]; dup {
				continue
			}
			seenObjects[object.Key] = struct{}{}

			documentName := filepath.Base(object.Key)
			documents = append(documents, StoredDocument{
				ObjectName:   object.Key,
				DocumentName: documentName,
				Size:         object.Size,
				LastModified: object.LastModified.Unix(),
				DocumentType: documentType(documentName),
			})
		}
	}

	s.logger.Info("Listed tender documents from storage", map[string]interface{}{
		"contract_folder_id": contractFolderID,
		"notice_id":          noticePublicationID,
		"prefixes":           prefixes,
		"documents_count":    len(documents),
	})

	return documents, nil
}

// isTenderDocumentObject reports whether an object key under the given prefix
// represents a downloadable tender document (PDF/DOCX/...) rather than an
// internal artefact such as tender.json or a cache file.
func isTenderDocumentObject(prefix, objectName string) bool {
	if objectName == "" || strings.HasSuffix(objectName, "/") {
		return false
	}

	relativePath := strings.TrimPrefix(objectName, prefix)
	if relativePath == "" || relativePath == objectName {
		// objectName did not start with prefix; ignore.
		return false
	}

	filename := strings.ToLower(filepath.Base(relativePath))
	switch filename {
	case "tender.json", "ai_cache.db":
		return false
	}

	switch strings.ToLower(filepath.Ext(filename)) {
	case ".json", ".db", ".sqlite", ".sqlite3":
		return false
	default:
		return true
	}
}

// DownloadDocument opens a document object from the configured MinIO bucket.
func (s *StorageClient) DownloadDocument(ctx context.Context, objectName string) (io.ReadCloser, error) {
	objectName = strings.TrimSpace(objectName)
	if objectName == "" {
		return nil, fmt.Errorf("ai_summarizer: objectName is required")
	}

	object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectName, minio.GetObjectOptions{})
	if err != nil {
		var minioErr minio.ErrorResponse
		if errors.As(err, &minioErr) && minioErr.StatusCode == http.StatusNotFound {
			return nil, ErrObjectNotFound
		}
		return nil, fmt.Errorf("ai_summarizer: failed to get document object: %w", err)
	}

	return object, nil
}

// GetAllSummaries walks the MinIO bucket for tender.json objects. It first lists
// under the PROC_ prefix; if none match, it falls back to a full-bucket recursive
// scan (slower on large buckets) so alternate key layouts still work.
func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryItem, error) {
	results := make([]TenderSummaryItem, 0)
	seen := make(map[string]struct{})

	appendFromPrefix := func(prefix string) {
		s.logger.Debug("Listing tender.json files from MinIO", map[string]interface{}{
			"bucket": s.config.MinioBucket,
			"prefix": prefix,
		})

		objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{
			Prefix:    prefix,
			Recursive: true,
		})

		for object := range objectCh {
			if object.Err != nil {
				s.logger.Error("Error listing MinIO objects", map[string]interface{}{
					"prefix": prefix,
					"error":  object.Err.Error(),
				})
				continue
			}
			key := object.Key
			if !strings.HasSuffix(key, "/tender.json") {
				continue
			}
			if _, dup := seen[key]; dup {
				continue
			}

			contractFolderID, noticePublicationID, ok := parseTenderJSONKey(key)
			if !ok {
				s.logger.Debug("Skipping unparseable tender.json key", map[string]interface{}{
					"key": key,
				})
				continue
			}
			seen[key] = struct{}{}

			summary, err := s.GetSummaryFromStorage(ctx, contractFolderID, noticePublicationID)
			if errors.Is(err, ErrObjectNotFound) {
				summary, err = s.getSummaryFromObjectKey(ctx, key, contractFolderID, noticePublicationID)
			}
			if err != nil {
				switch {
				case errors.Is(err, ErrSummaryNotReady):
					results = append(results, TenderSummaryItem{
						ContractFolderID: contractFolderID,
						NoticeID:         noticePublicationID,
						Done:             false,
					})
				case errors.Is(err, ErrNoOverallSummary):
					// File exists and pipeline reports done, but no summary text yet (or empty).
					results = append(results, TenderSummaryItem{
						ContractFolderID: contractFolderID,
						NoticeID:         noticePublicationID,
						OverallSummary:   "",
						Done:             true,
					})
				default:
					s.logger.Debug("Skipping tender summary", map[string]interface{}{
						"object_key":         key,
						"contract_folder_id": contractFolderID,
						"notice_id":          noticePublicationID,
						"error":              err.Error(),
					})
				}
				continue
			}

			results = append(results, TenderSummaryItem{
				ContractFolderID: contractFolderID,
				NoticeID:         noticePublicationID,
				OverallSummary:   summary,
				Done:             true,
			})
		}
	}

	appendFromPrefix(procedurePrefix)
	if len(seen) == 0 {
		s.logger.Info("No tender.json keys under PROC_; listing entire bucket for */tender.json (may be slower on large buckets)", map[string]interface{}{
			"bucket": s.config.MinioBucket,
		})
		appendFromPrefix("")
	}

	s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{
		"total_count":     len(results),
		"completed_count": countDone(results),
	})

	return results, nil
}

// parseTenderJSONKey splits an object key ending in /tender.json into contract folder
// id and notice publication id. Accepts PROC_/proc_ prefix or plain <folder>/<notice>/tender.json.
func parseTenderJSONKey(key string) (contractFolderID, noticePublicationID string, ok bool) {
	key = strings.TrimPrefix(strings.TrimSpace(key), "/")
	if !strings.HasSuffix(key, "/tender.json") {
		return "", "", false
	}
	trimmed := strings.TrimSuffix(key, "/tender.json")
	if trimmed == "" {
		return "", "", false
	}
	last := strings.LastIndex(trimmed, "/")
	if last < 0 {
		return "", "", false
	}
	noticePublicationID = strings.TrimSpace(trimmed[last+1:])
	parent := strings.TrimSpace(trimmed[:last])
	if noticePublicationID == "" || parent == "" {
		return "", "", false
	}
	contractFolderID = stripLeadingProcPrefix(parent)
	if contractFolderID == "" {
		return "", "", false
	}
	return contractFolderID, noticePublicationID, true
}

func stripLeadingProcPrefix(parent string) string {
	if len(parent) >= 5 && strings.EqualFold(parent[:5], "proc_") {
		return strings.TrimSpace(parent[5:])
	}
	return parent
}

// countDone counts how many items have Done=true
func countDone(items []TenderSummaryItem) int {
	count := 0
	for _, item := range items {
		if item.Done {
			count++
		}
	}
	return count
}

func documentType(filename string) string {
	extension := strings.TrimPrefix(strings.ToLower(filepath.Ext(filename)), ".")
	if extension == "" {
		return "unknown"
	}
	return extension
}