jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/newclean (pull request #81)

Browse Source 
Same Clean Output

* bitoftesting

* cleantest
This commit is contained in:
Michael McGuinness
2025-03-03 21:56:17 +00:00
parent 46dce9a64b
commit 2061543fc9
16 changed files with 646 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/docCleanRunner/runner_test.go
+16 -4
View File
@@ -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(
api/docTextRunner/runner_test.go
+3 -3
View File
@@ -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"}).
database/migrations/00000000000006_documents.up.sql
+21 -4
View File
@@ -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,
database/migrations/00000000000102_job_views.up.sql
+4 -3
View File
@@ -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,
database/queries/clean.sql
+25 -3
View File
@@ -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;
internal/database/repository/clean.sql.go
+95 -14
View File
@@ -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
internal/database/repository/clean_test.go
+57 -15
View File
@@ -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)
}
internal/database/repository/job_test.go
+24 -12
View File
@@ -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)
internal/database/repository/models.go
+55 -32
View File
@@ -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 {
internal/database/repository/result_test.go
+36 -20
View File
@@ -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)
internal/database/repository/text_test.go
+9 -5
View File
@@ -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)
internal/document/clean/clean.go
+68 -6
View File
@@ -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
}
internal/document/clean/clean_test.go
+211 -4
View File
@@ -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
internal/document/clean/create_test.go
+16 -4
View File
@@ -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(
internal/document/text/create_test.go
+3 -3
View File
@@ -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"}).
internal/document/text/extract_test.go
+3 -3
View File
@@ -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"}).