diff --git a/api/docCleanRunner/runner_test.go b/api/docCleanRunner/runner_test.go index d72006b6..bd740fc0 100644 --- a/api/docCleanRunner/runner_test.go +++ b/api/docCleanRunner/runner_test.go @@ -94,12 +94,24 @@ func TestDocCleanRunner(t *testing.T) { AddRow(dbid, inloc.Bucket, inloc.Key), ) mimeType := "application/pdf" - dbmimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.Cleanmimetypes(mimeType), + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), } - pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(dbid, int32(1), &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + pool.ExpectBegin() + cleanid := database.MustToDBUUID(uuid.New()) + pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid). + WillReturnRows( + pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}), + ) + pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(cleanid), + ) + pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)). WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() mockSQS.EXPECT(). SendMessage( diff --git a/api/docTextRunner/runner_test.go b/api/docTextRunner/runner_test.go index 40465755..79519738 100644 --- a/api/docTextRunner/runner_test.go +++ b/api/docTextRunner/runner_test.go @@ -72,9 +72,9 @@ func TestDocCleanRunner(t *testing.T) { ) cleanId := database.MustToDBUUID(uuid.New()) mimeType := "application/pdf" - dbmimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.Cleanmimetypes(mimeType), + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), } pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows( pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}). diff --git a/database/migrations/00000000000006_documents.up.sql b/database/migrations/00000000000006_documents.up.sql index 8cb1339f..edb66609 100644 --- a/database/migrations/00000000000006_documents.up.sql +++ b/database/migrations/00000000000006_documents.up.sql @@ -14,16 +14,26 @@ CREATE TABLE documentEntries ( foreign key (documentId) references documents(id) ); -CREATE TYPE cleanFailType AS ENUM ('invalid_mimetype'); -CREATE TYPE cleanMimeTypes AS ENUM ('application/pdf'); +CREATE TYPE cleanFailType AS ENUM ( + 'invalid_mimetype', + 'invalid_read', + 'invalid_read_pages', + 'zero_page_count', + 'large_page_count', + 'large_file', + 'small_dimensions', + 'large_dimensions', + 'small_dpi', + 'large_dpi' +); +CREATE TYPE cleanMimeType AS ENUM ('application/pdf'); CREATE TABLE documentCleans ( id uuid primary key DEFAULT uuid_generate_v7(), documentId uuid not null, - version int not null, bucket text, key text, - mimetype cleanMimeTypes, + mimetype cleanMimeType, fail cleanFailType, foreign key (documentId) references documents(id), CONSTRAINT bucket_and_key_together CHECK ((bucket IS NULL) = (key IS NULL) and (bucket is null) = (mimetype is null)), @@ -32,6 +42,13 @@ CREATE TABLE documentCleans ( ) ); +CREATE TABLE documentCleanEntries ( + id uuid primary key DEFAULT uuid_generate_v7(), + cleanId uuid not null, + version int not null, + foreign key (cleanId) references documentCleans(id) +); + CREATE TABLE documentTextExtractions ( id uuid primary key DEFAULT uuid_generate_v7(), cleanEntryId uuid not null, diff --git a/database/migrations/00000000000102_job_views.up.sql b/database/migrations/00000000000102_job_views.up.sql index 2baa48ac..057bdf90 100644 --- a/database/migrations/00000000000102_job_views.up.sql +++ b/database/migrations/00000000000102_job_views.up.sql @@ -26,14 +26,15 @@ WITH RankedExtractions AS ( dc.documentId, dc.bucket, dc.key, - dc.version, dc.mimetype, dc.fail, - ROW_NUMBER() OVER (PARTITION BY dc.documentId ORDER BY dc.id DESC) as row_num + dce.version, + ROW_NUMBER() OVER (PARTITION BY dc.documentId ORDER BY dc.id DESC, dce.id DESC) as row_num FROM documents d JOIN currentCollectorMinCleanVersions ccv ON d.jobId = ccv.jobId JOIN documentCleans dc ON dc.documentId = d.id - AND dc.version >= ccv.minCleanVersion + JOIN documentCleanEntries dce ON dc.id = dce.cleanId + AND dce.version >= ccv.minCleanVersion ) SELECT id, diff --git a/database/queries/clean.sql b/database/queries/clean.sql index a47c6d54..31155ca6 100644 --- a/database/queries/clean.sql +++ b/database/queries/clean.sql @@ -3,10 +3,32 @@ SELECT EXISTS( SELECT 1 FROM currentCleanEntries WHERE documentId = @documentId ); --- name: AddDocumentCleanEntry :exec -INSERT INTO documentCleans (documentId, version, bucket, key, mimetype, fail) VALUES ($1, $2, $3, $4, $5, $6); - -- name: GetDocumentCleanEntry :one SELECT id, documentId, bucket, key, version, mimetype, fail FROM currentCleanEntries WHERE documentId = @documentId; + +-- name: AddDocumentClean :one +INSERT INTO documentCleans (documentId, bucket, key, mimetype, fail) VALUES ($1, $2, $3, $4, $5) returning id; + +-- name: AddDocumentCleanEntry :exec +INSERT INTO documentCleanEntries (cleanId, version) VALUES ($1, $2); + +-- name: GetMostRecentDocumentCleanEntry :one +WITH cleans as ( + SELECT id, documentId, bucket, key, mimetype, fail + FROM documentCleans dc + WHERE documentId = @documentId ORDER BY id DESC +) +SELECT + dc.id, + dc.documentId, + dc.bucket, + dc.key, + dce.version, + dc.mimetype, + dc.fail +FROM cleans dc +JOIN documentCleanEntries dce ON dc.id = dce.cleanId +ORDER BY dc.id DESC, dce.id DESC +LIMIT 1; diff --git a/internal/database/repository/clean.sql.go b/internal/database/repository/clean.sql.go index 53a022c6..d7fdc45f 100644 --- a/internal/database/repository/clean.sql.go +++ b/internal/database/repository/clean.sql.go @@ -11,31 +11,48 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) -const addDocumentCleanEntry = `-- name: AddDocumentCleanEntry :exec -INSERT INTO documentCleans (documentId, version, bucket, key, mimetype, fail) VALUES ($1, $2, $3, $4, $5, $6) +const addDocumentClean = `-- name: AddDocumentClean :one +INSERT INTO documentCleans (documentId, bucket, key, mimetype, fail) VALUES ($1, $2, $3, $4, $5) returning id ` -type AddDocumentCleanEntryParams struct { - Documentid pgtype.UUID `db:"documentid"` - Version int32 `db:"version"` - Bucket *string `db:"bucket"` - Key *string `db:"key"` - Mimetype NullCleanmimetypes `db:"mimetype"` - Fail NullCleanfailtype `db:"fail"` +type AddDocumentCleanParams struct { + Documentid pgtype.UUID `db:"documentid"` + Bucket *string `db:"bucket"` + Key *string `db:"key"` + Mimetype NullCleanmimetype `db:"mimetype"` + Fail NullCleanfailtype `db:"fail"` } -// AddDocumentCleanEntry +// AddDocumentClean // -// INSERT INTO documentCleans (documentId, version, bucket, key, mimetype, fail) VALUES ($1, $2, $3, $4, $5, $6) -func (q *Queries) AddDocumentCleanEntry(ctx context.Context, arg *AddDocumentCleanEntryParams) error { - _, err := q.db.Exec(ctx, addDocumentCleanEntry, +// INSERT INTO documentCleans (documentId, bucket, key, mimetype, fail) VALUES ($1, $2, $3, $4, $5) returning id +func (q *Queries) AddDocumentClean(ctx context.Context, arg *AddDocumentCleanParams) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, addDocumentClean, arg.Documentid, - arg.Version, arg.Bucket, arg.Key, arg.Mimetype, arg.Fail, ) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} + +const addDocumentCleanEntry = `-- name: AddDocumentCleanEntry :exec +INSERT INTO documentCleanEntries (cleanId, version) VALUES ($1, $2) +` + +type AddDocumentCleanEntryParams struct { + Cleanid pgtype.UUID `db:"cleanid"` + Version int32 `db:"version"` +} + +// AddDocumentCleanEntry +// +// INSERT INTO documentCleanEntries (cleanId, version) VALUES ($1, $2) +func (q *Queries) AddDocumentCleanEntry(ctx context.Context, arg *AddDocumentCleanEntryParams) error { + _, err := q.db.Exec(ctx, addDocumentCleanEntry, arg.Cleanid, arg.Version) return err } @@ -65,6 +82,70 @@ func (q *Queries) GetDocumentCleanEntry(ctx context.Context, documentid pgtype.U return &i, err } +const getMostRecentDocumentCleanEntry = `-- name: GetMostRecentDocumentCleanEntry :one +WITH cleans as ( + SELECT id, documentId, bucket, key, mimetype, fail + FROM documentCleans dc + WHERE documentId = $1 ORDER BY id DESC +) +SELECT + dc.id, + dc.documentId, + dc.bucket, + dc.key, + dce.version, + dc.mimetype, + dc.fail +FROM cleans dc +JOIN documentCleanEntries dce ON dc.id = dce.cleanId +ORDER BY dc.id DESC, dce.id DESC +LIMIT 1 +` + +type GetMostRecentDocumentCleanEntryRow struct { + ID pgtype.UUID `db:"id"` + Documentid pgtype.UUID `db:"documentid"` + Bucket *string `db:"bucket"` + Key *string `db:"key"` + Version int32 `db:"version"` + Mimetype NullCleanmimetype `db:"mimetype"` + Fail NullCleanfailtype `db:"fail"` +} + +// GetMostRecentDocumentCleanEntry +// +// WITH cleans as ( +// SELECT id, documentId, bucket, key, mimetype, fail +// FROM documentCleans dc +// WHERE documentId = $1 ORDER BY id DESC +// ) +// SELECT +// dc.id, +// dc.documentId, +// dc.bucket, +// dc.key, +// dce.version, +// dc.mimetype, +// dc.fail +// FROM cleans dc +// JOIN documentCleanEntries dce ON dc.id = dce.cleanId +// ORDER BY dc.id DESC, dce.id DESC +// LIMIT 1 +func (q *Queries) GetMostRecentDocumentCleanEntry(ctx context.Context, documentid pgtype.UUID) (*GetMostRecentDocumentCleanEntryRow, error) { + row := q.db.QueryRow(ctx, getMostRecentDocumentCleanEntry, documentid) + var i GetMostRecentDocumentCleanEntryRow + err := row.Scan( + &i.ID, + &i.Documentid, + &i.Bucket, + &i.Key, + &i.Version, + &i.Mimetype, + &i.Fail, + ) + return &i, err +} + const hasDocumentCleanEntry = `-- name: HasDocumentCleanEntry :one SELECT EXISTS( SELECT 1 FROM currentCleanEntries WHERE documentId = $1 diff --git a/internal/database/repository/clean_test.go b/internal/database/repository/clean_test.go index a61c63ef..bf009602 100644 --- a/internal/database/repository/clean_test.go +++ b/internal/database/repository/clean_test.go @@ -9,7 +9,6 @@ import ( "queryorchestration/internal/test" "testing" - "github.com/jackc/pgx/v5/pgtype" "github.com/stretchr/testify/assert" ) @@ -52,17 +51,16 @@ func TestClean(t *testing.T) { bucket := "example_bucket" key := "example_key" - mimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + mimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, } fail := repository.NullCleanfailtype{ Valid: true, Cleanfailtype: repository.CleanfailtypeInvalidMimetype, } - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + _, err = queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: id, - Version: 1, Bucket: &bucket, Key: &key, Mimetype: mimetype, @@ -70,12 +68,16 @@ func TestClean(t *testing.T) { }) assert.Error(t, err) - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + failcleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: id, - Version: 1, Fail: fail, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Version: 1, + Cleanid: failcleanid, + }) + assert.NoError(t, err) isclean, err = queries.HasDocumentCleanEntry(ctx, id) assert.NoError(t, err) @@ -88,17 +90,20 @@ func TestClean(t *testing.T) { assert.Nil(t, clean.Key) assert.Equal(t, fail, clean.Fail) assert.Equal(t, int32(1), clean.Version) - assert.NotEqual(t, pgtype.UUID{}, clean.ID) - assert.True(t, clean.ID.Valid) + assert.Equal(t, failcleanid, clean.ID) - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: id, - Version: 2, Bucket: &bucket, Key: &key, Mimetype: mimetype, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Version: 2, + Cleanid: cleanid, + }) + assert.NoError(t, err) isclean, err = queries.HasDocumentCleanEntry(ctx, id) assert.NoError(t, err) @@ -107,11 +112,48 @@ func TestClean(t *testing.T) { clean, err = queries.GetDocumentCleanEntry(ctx, id) assert.NoError(t, err) assert.Equal(t, id, clean.Documentid) + assert.False(t, clean.Fail.Valid) assert.Equal(t, bucket, *clean.Bucket) assert.Equal(t, key, *clean.Key) assert.Equal(t, mimetype, clean.Mimetype) - assert.False(t, clean.Fail.Valid) assert.Equal(t, int32(2), clean.Version) - assert.NotEqual(t, pgtype.UUID{}, clean.ID) - assert.True(t, clean.ID.Valid) + assert.Equal(t, cleanid, clean.ID) + + recent, err := queries.GetMostRecentDocumentCleanEntry(ctx, id) + assert.NoError(t, err) + assert.Equal(t, clean.Documentid, recent.Documentid) + assert.Equal(t, clean.Bucket, recent.Bucket) + assert.Equal(t, clean.Key, recent.Key) + assert.Equal(t, clean.Mimetype, recent.Mimetype) + assert.Equal(t, clean.Fail, recent.Fail) + assert.Equal(t, clean.Version, recent.Version) + assert.Equal(t, clean.ID, recent.ID) + + version, err := queries.AddLatestCollectorVersion(ctx, jobId) + assert.NoError(t, err) + err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{ + Jobid: jobId, + Versionid: version, + }) + assert.NoError(t, err) + err = queries.SetCollectorCleanVersion(ctx, &repository.SetCollectorCleanVersionParams{ + Jobid: jobId, + Versionid: 5, + Addedversion: version, + }) + assert.NoError(t, err) + + isclean, err = queries.HasDocumentCleanEntry(ctx, id) + assert.NoError(t, err) + assert.False(t, isclean) + + recent, err = queries.GetMostRecentDocumentCleanEntry(ctx, id) + assert.NoError(t, err) + assert.Equal(t, clean.Documentid, recent.Documentid) + assert.Equal(t, clean.Bucket, recent.Bucket) + assert.Equal(t, clean.Key, recent.Key) + assert.Equal(t, clean.Mimetype, recent.Mimetype) + assert.Equal(t, clean.Fail, recent.Fail) + assert.Equal(t, clean.Version, recent.Version) + assert.Equal(t, clean.ID, recent.ID) } diff --git a/internal/database/repository/job_test.go b/internal/database/repository/job_test.go index 7cffec03..26456a04 100644 --- a/internal/database/repository/job_test.go +++ b/internal/database/repository/job_test.go @@ -243,15 +243,19 @@ func TestJobSync(t *testing.T) { bucket := "example_bucket" key := "example_key" - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentNoCleanID, - Version: 1, Fail: repository.NullCleanfailtype{ Valid: true, Cleanfailtype: repository.CleanfailtypeInvalidMimetype, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleanid, + Version: 1, + }) + assert.NoError(t, err) isSynced, err = queries.IsJobSynced(ctx, jobId) assert.NoError(t, err) @@ -268,17 +272,21 @@ func TestJobSync(t *testing.T) { assert.False(t, isSynced) keytwo := "example_key_2" - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleantwoid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentID, - Version: 1, Bucket: &bucket, Key: &keytwo, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleantwoid, + Version: 1, + }) + assert.NoError(t, err) isSynced, err = queries.IsJobSynced(ctx, jobId) assert.NoError(t, err) @@ -431,17 +439,21 @@ func TestJobSync(t *testing.T) { assert.NoError(t, err) assert.True(t, isSynced) - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanthreeid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentID, - Version: 1, Bucket: &bucket, Key: &key, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleanthreeid, + Version: 1, + }) + assert.NoError(t, err) isSynced, err = queries.IsJobSynced(ctx, jobId) assert.NoError(t, err) diff --git a/internal/database/repository/models.go b/internal/database/repository/models.go index d142d344..029f162d 100644 --- a/internal/database/repository/models.go +++ b/internal/database/repository/models.go @@ -14,7 +14,16 @@ import ( type Cleanfailtype string const ( - CleanfailtypeInvalidMimetype Cleanfailtype = "invalid_mimetype" + CleanfailtypeInvalidMimetype Cleanfailtype = "invalid_mimetype" + CleanfailtypeInvalidRead Cleanfailtype = "invalid_read" + CleanfailtypeInvalidReadPages Cleanfailtype = "invalid_read_pages" + CleanfailtypeZeroPageCount Cleanfailtype = "zero_page_count" + CleanfailtypeLargePageCount Cleanfailtype = "large_page_count" + CleanfailtypeLargeFile Cleanfailtype = "large_file" + CleanfailtypeSmallDimensions Cleanfailtype = "small_dimensions" + CleanfailtypeLargeDimensions Cleanfailtype = "large_dimensions" + CleanfailtypeSmallDpi Cleanfailtype = "small_dpi" + CleanfailtypeLargeDpi Cleanfailtype = "large_dpi" ) func (e *Cleanfailtype) Scan(src interface{}) error { @@ -54,56 +63,65 @@ func (ns NullCleanfailtype) Value() (driver.Value, error) { func (e Cleanfailtype) Valid() bool { switch e { - case CleanfailtypeInvalidMimetype: + case CleanfailtypeInvalidMimetype, + CleanfailtypeInvalidRead, + CleanfailtypeInvalidReadPages, + CleanfailtypeZeroPageCount, + CleanfailtypeLargePageCount, + CleanfailtypeLargeFile, + CleanfailtypeSmallDimensions, + CleanfailtypeLargeDimensions, + CleanfailtypeSmallDpi, + CleanfailtypeLargeDpi: return true } return false } -type Cleanmimetypes string +type Cleanmimetype string const ( - CleanmimetypesApplicationPdf Cleanmimetypes = "application/pdf" + CleanmimetypeApplicationPdf Cleanmimetype = "application/pdf" ) -func (e *Cleanmimetypes) Scan(src interface{}) error { +func (e *Cleanmimetype) Scan(src interface{}) error { switch s := src.(type) { case []byte: - *e = Cleanmimetypes(s) + *e = Cleanmimetype(s) case string: - *e = Cleanmimetypes(s) + *e = Cleanmimetype(s) default: - return fmt.Errorf("unsupported scan type for Cleanmimetypes: %T", src) + return fmt.Errorf("unsupported scan type for Cleanmimetype: %T", src) } return nil } -type NullCleanmimetypes struct { - Cleanmimetypes Cleanmimetypes - Valid bool // Valid is true if Cleanmimetypes is not NULL +type NullCleanmimetype struct { + Cleanmimetype Cleanmimetype + Valid bool // Valid is true if Cleanmimetype is not NULL } // Scan implements the Scanner interface. -func (ns *NullCleanmimetypes) Scan(value interface{}) error { +func (ns *NullCleanmimetype) Scan(value interface{}) error { if value == nil { - ns.Cleanmimetypes, ns.Valid = "", false + ns.Cleanmimetype, ns.Valid = "", false return nil } ns.Valid = true - return ns.Cleanmimetypes.Scan(value) + return ns.Cleanmimetype.Scan(value) } // Value implements the driver Valuer interface. -func (ns NullCleanmimetypes) Value() (driver.Value, error) { +func (ns NullCleanmimetype) Value() (driver.Value, error) { if !ns.Valid { return nil, nil } - return string(ns.Cleanmimetypes), nil + return string(ns.Cleanmimetype), nil } -func (e Cleanmimetypes) Valid() bool { +func (e Cleanmimetype) Valid() bool { switch e { - case CleanmimetypesApplicationPdf: + case CleanmimetypeApplicationPdf: return true } return false @@ -227,13 +245,13 @@ type Collectorversion struct { } type Currentcleanentry struct { - ID pgtype.UUID `db:"id"` - Documentid pgtype.UUID `db:"documentid"` - Bucket *string `db:"bucket"` - Key *string `db:"key"` - Version int32 `db:"version"` - Mimetype NullCleanmimetypes `db:"mimetype"` - Fail NullCleanfailtype `db:"fail"` + ID pgtype.UUID `db:"id"` + Documentid pgtype.UUID `db:"documentid"` + Bucket *string `db:"bucket"` + Key *string `db:"key"` + Version int32 `db:"version"` + Mimetype NullCleanmimetype `db:"mimetype"` + Fail NullCleanfailtype `db:"fail"` } type Currentclientcansync struct { @@ -283,13 +301,18 @@ type Document struct { } type Documentclean struct { - ID pgtype.UUID `db:"id"` - Documentid pgtype.UUID `db:"documentid"` - Version int32 `db:"version"` - Bucket *string `db:"bucket"` - Key *string `db:"key"` - Mimetype NullCleanmimetypes `db:"mimetype"` - Fail NullCleanfailtype `db:"fail"` + ID pgtype.UUID `db:"id"` + Documentid pgtype.UUID `db:"documentid"` + Bucket *string `db:"bucket"` + Key *string `db:"key"` + Mimetype NullCleanmimetype `db:"mimetype"` + Fail NullCleanfailtype `db:"fail"` +} + +type Documentcleanentry struct { + ID pgtype.UUID `db:"id"` + Cleanid pgtype.UUID `db:"cleanid"` + Version int32 `db:"version"` } type Documententry struct { diff --git a/internal/database/repository/result_test.go b/internal/database/repository/result_test.go index 4b95a43f..a072571f 100644 --- a/internal/database/repository/result_test.go +++ b/internal/database/repository/result_test.go @@ -73,17 +73,21 @@ func TestResults(t *testing.T) { bucket := "example_bucket" key := "example_key" - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentID, - Version: 1, Bucket: &bucket, Key: &key, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleanid, + Version: 1, + }) + assert.NoError(t, err) issynced, err = queries.IsJobSynced(ctx, jobId) assert.NoError(t, err) @@ -235,17 +239,21 @@ func TestResultValues(t *testing.T) { bucket := "example_bucket" key := "example_key" - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentID, - Version: 1, Bucket: &bucket, Key: &key, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleanid, + Version: 1, + }) + assert.NoError(t, err) cleanentry, err := queries.GetDocumentCleanEntry(ctx, documentID) assert.NoError(t, err) err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{ @@ -433,17 +441,21 @@ func TestUnsyncedNoDepsQueries(t *testing.T) { bucket := "example_bucket" key := "example_key" - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentID, - Version: 1, Bucket: &bucket, Key: &key, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleanid, + Version: 1, + }) + assert.NoError(t, err) qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, documentID) assert.NoError(t, err) @@ -505,17 +517,21 @@ func TestUnsyncedNoDepsQueries(t *testing.T) { assert.NoError(t, err) assert.Len(t, qs, 0) - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleantwoid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: documentTwoID, - Version: 1, Bucket: &bucket, Key: &key, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleantwoid, + Version: 1, + }) + assert.NoError(t, err) qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, documentID) assert.NoError(t, err) diff --git a/internal/database/repository/text_test.go b/internal/database/repository/text_test.go index 6712ec54..08b623a0 100644 --- a/internal/database/repository/text_test.go +++ b/internal/database/repository/text_test.go @@ -45,17 +45,21 @@ func TestTextExtraction(t *testing.T) { bucket := "example_bucket" key := "example_key" - err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{ Documentid: id, - Version: 1, Bucket: &bucket, Key: &key, - Mimetype: repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.CleanmimetypesApplicationPdf, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.CleanmimetypeApplicationPdf, }, }) assert.NoError(t, err) + err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: cleanid, + Version: 1, + }) + assert.NoError(t, err) cleanentry, err := queries.GetDocumentCleanEntry(ctx, id) assert.NoError(t, err) diff --git a/internal/document/clean/clean.go b/internal/document/clean/clean.go index 01f70cf7..c936e4f4 100644 --- a/internal/document/clean/clean.go +++ b/internal/document/clean/clean.go @@ -2,6 +2,8 @@ package documentclean import ( "context" + "database/sql" + "errors" "fmt" "log/slog" "queryorchestration/internal/database" @@ -108,11 +110,20 @@ func (s *Service) clean(ctx context.Context, id uuid.UUID) error { return err } + err = s.storeClean(ctx, id, out) + if err != nil { + return err + } + + return nil +} + +func (s *Service) storeClean(ctx context.Context, id uuid.UUID, out *ExecuteCleanResponse) error { + docId := database.MustToDBUUID(id) version := s.svc.Version.GetVersion() - params := &repository.AddDocumentCleanEntryParams{ + params := &repository.AddDocumentCleanParams{ Documentid: docId, - Version: version, } if out.failReason != nil { slog.Info("Failed document", "id", id, "reason", *out.failReason) @@ -123,13 +134,38 @@ func (s *Service) clean(ctx context.Context, id uuid.UUID) error { } else { params.Bucket = &out.location.Bucket params.Key = &out.location.Key - params.Mimetype = repository.NullCleanmimetypes{ - Cleanmimetypes: repository.Cleanmimetypes(*out.mimetype), - Valid: true, + params.Mimetype = repository.NullCleanmimetype{ + Cleanmimetype: repository.Cleanmimetype(*out.mimetype), + Valid: true, } } - err = s.cfg.GetDBQueries().AddDocumentCleanEntry(ctx, params) + err := s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, q *repository.Queries) error { + lastClean, err := q.GetMostRecentDocumentCleanEntry(ctx, docId) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + + newID := lastClean.ID + if s.isNewClean(lastClean, out) { + id, err := q.AddDocumentClean(ctx, params) + if err != nil { + return err + } + + newID = id + } + + err = q.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ + Cleanid: newID, + Version: version, + }) + if err != nil { + return err + } + + return nil + }) if err != nil { return err } @@ -140,3 +176,29 @@ func (s *Service) clean(ctx context.Context, id uuid.UUID) error { return nil } + +func (s *Service) isNewClean(lastEntry *repository.GetMostRecentDocumentCleanEntryRow, out *ExecuteCleanResponse) bool { + if lastEntry == nil || !lastEntry.ID.Valid { + return true + } + + if out.failReason != nil { + if lastEntry.Fail.Valid && *out.failReason == InvalidDocumentReason(lastEntry.Fail.Cleanfailtype) { + return false + } + + return true + } + + if lastEntry.Fail.Valid { + return true + } + + if out.location.Bucket == *lastEntry.Bucket && + out.location.Key == *lastEntry.Key && + *out.mimetype == MimeType(lastEntry.Mimetype.Cleanmimetype) { + return false + } + + return true +} diff --git a/internal/document/clean/clean_test.go b/internal/document/clean/clean_test.go index 48ddd2f8..58e6d82e 100644 --- a/internal/document/clean/clean_test.go +++ b/internal/document/clean/clean_test.go @@ -59,12 +59,24 @@ func TestClean(t *testing.T) { AddRow(dbid, inloc.Bucket, inloc.Key), ) mimeType := "application/pdf" - dbmimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.Cleanmimetypes(mimeType), + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), } - pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(dbid, int32(1), &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + pool.ExpectBegin() + pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid). + WillReturnRows( + pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}), + ) + cleanid := database.MustToDBUUID(uuid.New()) + pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(cleanid), + ) + pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)). WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() var length int64 = 10 mockS3.EXPECT(). @@ -151,6 +163,201 @@ func TestExecuteCleanTasks(t *testing.T) { }, *outloc) } +func TestStoreClean(t *testing.T) { + ctx := context.Background() + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + cfg := &DocCleanConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + svc := Service{ + cfg: cfg, + svc: &Services{ + Version: cleanversion.New(cfg), + }, + } + + doc := document.Document{ + ID: uuid.New(), + } + inloc := document.Location{ + Bucket: "bucket_name", + Key: "/i/am/here", + } + t.Run("no prev entry", func(t *testing.T) { + mimeType := "application/pdf" + params := &ExecuteCleanResponse{ + location: &inloc, + mimetype: (*MimeType)(&mimeType), + } + dbid := database.MustToDBUUID(doc.ID) + + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), + } + pool.ExpectBegin() + pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid). + WillReturnRows( + pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}), + ) + cleanid := database.MustToDBUUID(uuid.New()) + pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(cleanid), + ) + pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + err = svc.storeClean(ctx, doc.ID, params) + assert.NoError(t, err) + }) + t.Run("diff prev entry", func(t *testing.T) { + mimeType := "application/pdf" + params := &ExecuteCleanResponse{ + location: &inloc, + mimetype: (*MimeType)(&mimeType), + } + dbid := database.MustToDBUUID(doc.ID) + + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), + } + pool.ExpectBegin() + pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid). + WillReturnRows( + pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}). + AddRow(database.MustToDBUUID(uuid.New()), dbid, nil, nil, int32(1), nil, repository.CleanfailtypeInvalidRead), + ) + cleanid := database.MustToDBUUID(uuid.New()) + pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(cleanid), + ) + pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + err = svc.storeClean(ctx, doc.ID, params) + assert.NoError(t, err) + }) + t.Run("same fail", func(t *testing.T) { + reason := InvalidDocumentMimeType + params := &ExecuteCleanResponse{ + failReason: &reason, + } + dbid := database.MustToDBUUID(doc.ID) + cleanid := database.MustToDBUUID(uuid.New()) + + pool.ExpectBegin() + pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid). + WillReturnRows( + pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}). + AddRow(cleanid, dbid, nil, nil, int32(1), nil, repository.NullCleanfailtype{ + Valid: true, + Cleanfailtype: repository.CleanfailtypeInvalidMimetype, + }), + ) + pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + err = svc.storeClean(ctx, doc.ID, params) + assert.EqualError(t, err, "invalid_mimetype") + }) +} + +func TestIsNewClean(t *testing.T) { + svc := Service{} + t.Run("no last entry", func(t *testing.T) { + var lastEntry *repository.GetMostRecentDocumentCleanEntryRow = nil + params := &ExecuteCleanResponse{} + assert.True(t, svc.isNewClean(lastEntry, params)) + }) + t.Run("same fail", func(t *testing.T) { + var lastEntry *repository.GetMostRecentDocumentCleanEntryRow = &repository.GetMostRecentDocumentCleanEntryRow{ + ID: database.MustToDBUUID(uuid.New()), + Fail: repository.NullCleanfailtype{ + Valid: true, + Cleanfailtype: repository.CleanfailtypeInvalidMimetype, + }, + } + reason := InvalidDocumentMimeType + params := &ExecuteCleanResponse{ + failReason: &reason, + } + assert.False(t, svc.isNewClean(lastEntry, params)) + }) + t.Run("different fail", func(t *testing.T) { + var lastEntry *repository.GetMostRecentDocumentCleanEntryRow = &repository.GetMostRecentDocumentCleanEntryRow{ + ID: database.MustToDBUUID(uuid.New()), + Fail: repository.NullCleanfailtype{ + Valid: true, + Cleanfailtype: repository.CleanfailtypeInvalidMimetype, + }, + } + reason := InvalidDocumentLargeDPI + params := &ExecuteCleanResponse{ + failReason: &reason, + } + assert.True(t, svc.isNewClean(lastEntry, params)) + }) + t.Run("same location", func(t *testing.T) { + loc := document.Location{ + Bucket: "bucket", + Key: "hi", + } + mimeType := MimeTypePDF + var lastEntry *repository.GetMostRecentDocumentCleanEntryRow = &repository.GetMostRecentDocumentCleanEntryRow{ + ID: database.MustToDBUUID(uuid.New()), + Bucket: &loc.Bucket, + Key: &loc.Key, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), + }, + } + params := &ExecuteCleanResponse{ + location: &loc, + mimetype: &mimeType, + } + assert.False(t, svc.isNewClean(lastEntry, params)) + }) + t.Run("different location", func(t *testing.T) { + ogloc := document.Location{ + Bucket: "bucket", + Key: "hi", + } + newloc := document.Location{ + Bucket: "bucket", + Key: "different_location", + } + mimeType := MimeTypePDF + var lastEntry *repository.GetMostRecentDocumentCleanEntryRow = &repository.GetMostRecentDocumentCleanEntryRow{ + ID: database.MustToDBUUID(uuid.New()), + Bucket: &ogloc.Bucket, + Key: &ogloc.Key, + Mimetype: repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), + }, + } + params := &ExecuteCleanResponse{ + location: &newloc, + mimetype: &mimeType, + } + assert.True(t, svc.isNewClean(lastEntry, params)) + }) +} + const pdfHelloWorld = `%PDF-1.4 %���� 1 0 obj diff --git a/internal/document/clean/create_test.go b/internal/document/clean/create_test.go index 837b16f0..fe8376b5 100644 --- a/internal/document/clean/create_test.go +++ b/internal/document/clean/create_test.go @@ -79,12 +79,24 @@ func TestCreate(t *testing.T) { AddRow(dbid, inloc.Bucket, inloc.Key), ) mimeType := "application/pdf" - dbmimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.Cleanmimetypes(mimeType), + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), } - pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(dbid, int32(1), &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + pool.ExpectBegin() + pool.ExpectQuery("name: GetMostRecentDocumentCleanEntry :one").WithArgs(dbid). + WillReturnRows( + pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}), + ) + cleanid := database.MustToDBUUID(uuid.New()) + pool.ExpectQuery("name: AddDocumentClean :one").WithArgs(dbid, &inloc.Bucket, &inloc.Key, dbmimetype, repository.NullCleanfailtype{}). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(cleanid), + ) + pool.ExpectExec("name: AddDocumentCleanEntry :exec").WithArgs(cleanid, int32(1)). WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() mockSQS.EXPECT(). SendMessage( diff --git a/internal/document/text/create_test.go b/internal/document/text/create_test.go index 8b71267f..f4a60ede 100644 --- a/internal/document/text/create_test.go +++ b/internal/document/text/create_test.go @@ -58,9 +58,9 @@ func TestCreate(t *testing.T) { ) cleanId := database.MustToDBUUID(uuid.New()) mimeType := "application/pdf" - dbmimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.Cleanmimetypes(mimeType), + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), } pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}). diff --git a/internal/document/text/extract_test.go b/internal/document/text/extract_test.go index c5eec9f7..b774da9c 100644 --- a/internal/document/text/extract_test.go +++ b/internal/document/text/extract_test.go @@ -39,9 +39,9 @@ func TestExtract(t *testing.T) { cleanId := database.MustToDBUUID(uuid.New()) mimeType := "application/pdf" - dbmimetype := repository.NullCleanmimetypes{ - Valid: true, - Cleanmimetypes: repository.Cleanmimetypes(mimeType), + dbmimetype := repository.NullCleanmimetype{ + Valid: true, + Cleanmimetype: repository.Cleanmimetype(mimeType), } pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( pgxmock.NewRows([]string{"id", "documentId", "bucket", "key", "version", "mimetype", "fail"}).