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: ¶ms.Bucket, Key: &keyStr, IfMatch: ¶ms.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: ¶ms.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 }