query-orchestration/internal/document/text/processExtract.go

package documenttext

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"io"
	"strings"
	"time"

	"queryorchestration/internal/database/repository"
	"queryorchestration/internal/serviceconfig/build"
	"queryorchestration/internal/serviceconfig/objectstore"

	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/google/uuid"
)

type ProcessParams struct {
	Bucket string
	Key    objectstore.BucketKey
	Hash   string
}

type Page string

func (s *Service) getPages(baseTextract io.Reader) ([]*Page, error) {
	// For each page
	// Look for table, form, signature
	// If one/many found - run appropriate tier and merge in
	// tiers?? - analyze (for tables?)

	// cleanPages
	// linkPages

	return []*Page{}, nil
}

func (s *Service) mergePages(pages []*Page) (io.Reader, error) {
	// merge document - pages to text
	// e.g. common tables
	return strings.NewReader("hello"), nil
}

func (s *Service) processTrigger(ctx context.Context, trigger *repository.GetDocumentTextTriggerRow, params *ProcessParams) error {
	keyStr := params.Key.String()
	response, err := s.cfg.GetStoreClient().GetObject(ctx, &s3.GetObjectInput{
		Bucket:  &params.Bucket,
		Key:     &keyStr,
		IfMatch: &params.Hash,
	})
	if err != nil {
		return err
	}

	pages, err := s.getPages(response.Body)
	if err != nil {
		return err
	}

	text, err := s.mergePages(pages)
	if err != nil {
		return err
	}

	err = s.storeProcess(ctx, text, trigger, params)
	if err != nil {
		return err
	}

	return nil
}

func (s *Service) storeProcess(ctx context.Context, text io.Reader, trigger *repository.GetDocumentTextTriggerRow, params *ProcessParams) error {
	version := build.GetVersionUnixTimestamp()

	hashReader, storeReader, err := duplicateReader(text)
	if err != nil {
		return err
	}

	hash, err := s.cfg.CalculateETag(ctx, hashReader)
	if err != nil {
		return err
	}

	return s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, q *repository.Queries) error {
		existingId, err := q.GetDocumentTextExtractionByHash(ctx, &repository.GetDocumentTextExtractionByHashParams{
			Cleanentryid: trigger.Cleanid,
			Hash:         hash,
		})
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return err
		}

		var textId uuid.UUID
		if existingId == uuid.Nil || errors.Is(err, sql.ErrNoRows) {
			createdAt := time.Now().UTC()
			key := objectstore.BucketKey{
				ClientID:  params.Key.ClientID,
				Location:  objectstore.TextOut,
				CreatedAt: createdAt,
				EntityID:  trigger.ID,
			}
			keyStr := key.String()

			textId, err = q.AddDocumentText(ctx, &repository.AddDocumentTextParams{
				Bucket: params.Bucket,
				Key:    keyStr,
				Hash:   hash,
			})
			if err != nil {
				return err
			}

			_, err = s.cfg.GetStoreClient().PutObject(ctx, &s3.PutObjectInput{
				Bucket: &params.Bucket,
				Key:    &keyStr,
				Body:   storeReader,
			})
			if err != nil {
				return err
			}

		} else {
			textId = existingId
		}

		err = q.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
			Textid:    textId,
			Triggerid: trigger.ID,
			Version:   version,
		})
		if err != nil {
			return err
		}

		return nil
	})
}

func (s *Service) Process(ctx context.Context, params *ProcessParams) error {
	trigger, err := s.cfg.GetDBQueries().GetDocumentTextTrigger(ctx, params.Key.EntityID)
	if err != nil {
		return err
	}

	err = s.processTrigger(ctx, trigger, params)
	if err != nil {
		return err
	}

	err = s.informExtraction(ctx, trigger.Documentid)
	if err != nil {
		return err
	}

	return nil
}

func duplicateReader(reader io.Reader) (io.Reader, io.Reader, error) {
	content, err := io.ReadAll(reader)
	if err != nil {
		return nil, nil, err
	}

	reader1 := bytes.NewReader(content)
	reader2 := bytes.NewReader(content)

	return reader1, reader2, nil
}