Gleipnir/nidus-sync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Overhaul publicreport storage layer, create unified tables

Browse source 
This is a huge change. I was getting really sick of the split between
nuisance/water tables when more than half of the data they store is
common. I finally bit off the big work of switching it all.

This creates a single unified table, publicreport.report and copies the
existing report data into it. It also ports existing data from the
original tables into the new table.

Along with all of this I also overhauled the system for handling
asynchronous work to use a LISTEN/NOTIFY connection from the database
and a single cache table to avoid ever losing work.
This commit is contained in:
Eli Ribble 2026-03-18 15:36:20 +00:00
parent 2538638c9d
commit 1e071d5ce5
No known key found for this signature in database
109 changed files with 22903 additions and 11713 deletions
Download patch file Download diff file Expand all files Collapse all files

9
platform/background/background.go
View file

@ -3,6 +3,7 @@ package background
import (
"context"
"fmt"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
@ -26,11 +27,15 @@ func NewEmailSend(ctx context.Context, txn bob.Executor, email_id int32) error {
func NewLabelStudioAudioCreate(ctx context.Context, txn bob.Executor, note_audio_id int32) error {
return newJob(ctx, txn, enums.JobtypeLabelStudioAudioCreate, note_audio_id)
}
func NewTextSend(ctx context.Context, txn bob.Executor, text_id int32) error {
return newJob(ctx, txn, enums.JobtypeTextSend, text_id)
func NewTextRespond(ctx context.Context, txn bob.Executor, text_id int32) error {
return newJob(ctx, txn, enums.JobtypeTextRespond, text_id)
}
func NewTextSend(ctx context.Context, txn bob.Executor, job_id int32) error {
return newJob(ctx, txn, enums.JobtypeTextSend, job_id)
}
func newJob(ctx context.Context, txn bob.Executor, t enums.Jobtype, id int32) error {
_, err := models.Jobs.Insert(&models.JobSetter{
Created: omit.From(time.Now()),
// ID
Type: omit.From(t),
RowID: omit.From(id),

13
platform/communication.go
View file

@ -9,16 +9,9 @@ import (
)
func NotificationCount(ctx context.Context, org *models.Organization, user *models.User) (result uint, err error) {
count_nreports, err := publicreport.NuisanceReportForOrganizationCount(ctx, org.ID)
count_reports, err := publicreport.ReportsForOrganizationCount(ctx, org.ID)
if err != nil {
return 0, fmt.Errorf("nuisance report query: %w", err)
return 0, fmt.Errorf("report query: %w", err)
}
result += count_nreports
count_wreports, err := publicreport.WaterReportForOrganizationCount(ctx, org.ID)
if err != nil {
return 0, fmt.Errorf("water report query: %w", err)
}
result += count_wreports
return result, nil
return uint(count_reports), nil
}

4
platform/csv/flyover.go
View file

@ -225,8 +225,8 @@ func insertFlyover(ctx context.Context, txn bob.Tx, file *models.FileuploadFile,
um.SetCol("geom").To(geom_query),
um.SetCol("is_in_district").To(psql.F("ST_Contains", "org.service_area_geometry", geom_query)),
um.From("fileupload.csv").As("csv"),
um.InnerJoin("fileupload.file").As("file").OnEQ(psql.Raw("csv.file_id"), psql.Raw("file.id")),
um.InnerJoin("organization").As("org").OnEQ(psql.Raw("file.organization_id"), psql.Raw("org.id")),
um.InnerJoin("fileupload.file").As("file").OnEQ(psql.Quote("csv", "file_id"), psql.Quote("file", "id")),
um.InnerJoin("organization").As("org").OnEQ(psql.Quote("file", "organization_id"), psql.Quote("org", "id")),
um.Where(psql.Quote("pool", "id").EQ(psql.Arg(flyover.ID))),
).Exec(ctx, txn)
if err != nil {

1
platform/event/event.go
View file

@ -73,6 +73,7 @@ type ResourceType int
const (
TypeUnknown = iota
TypeRMONuisance
TypeRMOReport
TypeRMOWater
)

37
platform/label_studio.go
View file

@ -21,24 +21,27 @@ var labelStudioProject *labelstudio.Project
var minioClient *minio.Client
func initializeLabelStudio() error {
// Initialize the minio client
//minioBucket := os.Getenv("S3_BUCKET")
var err error
labelStudioClient, err = createLabelStudioClient()
if err != nil {
return fmt.Errorf("Failed to create label studio client: %w", err)
}
// Get the project we are going to upload to
labelStudioProject, err = findLabelStudioProject(labelStudioClient, "Nidus Speech-to-Text Transcriptions")
if err != nil {
return fmt.Errorf("Failed to find the label studio project: %w", err)
}
minioClient, err = createMinioClient()
if err != nil {
return fmt.Errorf("Failed to create minio client: %w", err)
}
return nil
/*
// Initialize the minio client
//minioBucket := os.Getenv("S3_BUCKET")
var err error
labelStudioClient, err = createLabelStudioClient()
if err != nil {
return fmt.Errorf("Failed to create label studio client: %w", err)
}
// Get the project we are going to upload to
labelStudioProject, err = findLabelStudioProject(labelStudioClient, "Nidus Speech-to-Text Transcriptions")
if err != nil {
return fmt.Errorf("Failed to find the label studio project: %w", err)
}
minioClient, err = createMinioClient()
if err != nil {
return fmt.Errorf("Failed to create minio client: %w", err)
}
return nil
*/
}
func createMinioClient() (*minio.Client, error) {
baseUrl := os.Getenv("S3_BASE_URL")

14
platform/notification.go
View file

@ -111,21 +111,15 @@ func NotificationCountsForUser(ctx context.Context, u User) (*UserNotificationCo
if err != nil {
return nil, fmt.Errorf("Failed to get home notification count: %w", err)
}
count_nuisance, err := u.Organization.model.Nuisances(
models.SelectWhere.PublicreportNuisances.Reviewed.IsNull(),
count_reports, err := u.Organization.model.Reports(
models.SelectWhere.PublicreportReports.Reviewed.IsNull(),
).Count(ctx, db.PGInstance.BobDB)
if err != nil {
return nil, fmt.Errorf("Failed to get nuisance notification count: %w", err)
}
count_water, err := u.Organization.model.Waters(
models.SelectWhere.PublicreportWaters.Reviewed.IsNull(),
).Count(ctx, db.PGInstance.BobDB)
if err != nil {
return nil, fmt.Errorf("Failed to get water notification count: %w", err)
}
log.Debug().Int64("nuisance", count_nuisance).Int64("water", count_water).Int64("home", count_home).Int("user", u.ID).Msg("calculated notification counts")
log.Debug().Int64("reports", count_reports).Int64("home", count_home).Int("user", u.ID).Msg("calculated notification counts")
return &UserNotificationCounts{
Communications: uint(count_nuisance + count_water),
Communications: uint(count_reports),
Home: uint(count_home),
}, nil
}

103
platform/nuisance.go
View file

@ -1,104 +1 @@
package platform
import (
"context"
"fmt"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/event"
"github.com/Gleipnir-Technology/nidus-sync/platform/geocode"
"github.com/Gleipnir-Technology/nidus-sync/platform/report"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
"github.com/rs/zerolog/log"
)
func NuisanceCreate(ctx context.Context, setter models.PublicreportNuisanceSetter, latlng LatLng, address Address, images []ImageUpload) (public_id string, err error) {
txn, err := db.PGInstance.BobDB.BeginTx(ctx, nil)
if err != nil {
return "", fmt.Errorf("create txn: %w", err)
}
defer txn.Rollback(ctx)
public_id, err = report.GenerateReportID()
if err != nil {
return "", fmt.Errorf("create public ID: %w", err)
}
setter.PublicID = omit.From(public_id)
// If we've got an locality value it was set by geocoding so we should save it
var a *models.Address
if address.Locality != "" && latlng.Latitude != nil && latlng.Longitude != nil {
a, err = geocode.EnsureAddress(ctx, txn, address, types.Location{
Latitude: *latlng.Latitude,
Longitude: *latlng.Longitude,
})
if err != nil {
return "", fmt.Errorf("Failed to ensure address: %w", err)
}
}
saved_images, err := saveImageUploads(ctx, txn, images)
if err != nil {
return "", fmt.Errorf("Failed to save image uploads: %w", err)
}
var organization_id *int32
organization_id, err = MatchDistrict(ctx, latlng.Longitude, latlng.Latitude, images)
if err != nil {
log.Warn().Err(err).Msg("Failed to match district")
}
if a != nil {
setter.AddressID = omitnull.From(a.ID)
}
if organization_id != nil {
setter.OrganizationID = omit.FromPtr(organization_id)
}
nuisance, err := models.PublicreportNuisances.Insert(&setter).One(ctx, txn)
if err != nil {
return "", fmt.Errorf("Failed to create database record: %w", err)
}
if latlng.Latitude != nil && latlng.Longitude != nil {
h3cell, _ := latlng.H3Cell()
geom_query, _ := latlng.GeometryQuery()
_, err = psql.Update(
um.Table("publicreport.nuisance"),
um.SetCol("h3cell").ToArg(h3cell),
um.SetCol("location").To(geom_query),
um.Where(psql.Quote("id").EQ(psql.Arg(nuisance.ID))),
).Exec(ctx, txn)
if err != nil {
return "", fmt.Errorf("Failed to insert publicreport.nuisance geospatial", err)
}
}
log.Info().Str("public_id", public_id).Int32("id", nuisance.ID).Msg("Created nuisance report")
if len(saved_images) > 0 {
setters := make([]*models.PublicreportNuisanceImageSetter, 0)
for _, image := range saved_images {
setters = append(setters, &models.PublicreportNuisanceImageSetter{
ImageID: omit.From(int32(image.ID)),
NuisanceID: omit.From(int32(nuisance.ID)),
})
}
_, err = models.PublicreportNuisanceImages.Insert(bob.ToMods(setters...)).Exec(ctx, txn)
if err != nil {
return "", fmt.Errorf("Failed to save reference to images: %w", err)
}
log.Info().Int("len", len(images)).Msg("saved uploaded images")
}
txn.Commit(ctx)
if organization_id != nil {
event.Created(
event.TypeRMONuisance,
*organization_id,
nuisance.PublicID,
)
}
return nuisance.PublicID, nil
}

77
platform/publicreport.go
View file

@ -6,9 +6,9 @@ import (
"fmt"
"time"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
//"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
@ -20,71 +20,68 @@ import (
)
func PublicreportInvalid(ctx context.Context, user User, report_id string) error {
tablename, _, err := reportFromID(ctx, user, report_id)
report, err := reportFromID(ctx, user, report_id)
if err != nil {
return fmt.Errorf("query report existence: %w", err)
}
_, err = psql.Update(
um.Table("publicreport."+tablename),
um.SetCol("reviewed").ToArg(time.Now()),
um.SetCol("reviewer_id").ToArg(user.ID),
um.SetCol("status").ToArg(enums.PublicreportReportstatustypeInvalidated),
um.Where(psql.Quote("public_id").EQ(psql.Arg(report_id))),
).Exec(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("update report %s.%s: %w", tablename, report_id, err)
}
err = report.Update(ctx, db.PGInstance.BobDB, &models.PublicreportReportSetter{
Reviewed: omitnull.From(time.Now()),
ReviewerID: omitnull.From(int32(user.ID)),
Status: omit.From(enums.PublicreportReportstatustypeInvalidated),
})
log.Info().Str("report-id", report_id).Str("tablename", tablename).Msg("Marked as invalid")
resource := resourceTypeFromTablename(tablename)
event.Updated(resource, user.Organization.ID(), report_id)
log.Info().Int32("id", report.ID).Msg("Report marked as invalid")
event.Updated(event.TypeRMOReport, user.Organization.ID(), report_id)
return nil
}
func PublicReportMessageCreate(ctx context.Context, user User, report_id, message string) (message_id *int32, err error) {
_, report, err := reportFromID(ctx, user, report_id)
txn, err := db.PGInstance.BobDB.BeginTx(ctx, nil)
if err != nil {
return nil, fmt.Errorf("create txn: %w", err)
}
defer txn.Rollback(ctx)
report, err := reportFromID(ctx, user, report_id)
if err != nil {
return nil, fmt.Errorf("query report existence: %w", err)
}
if report.ReporterPhone.GetOr("") != "" {
msg_id, err := text.ReportMessage(ctx, int32(user.ID), report_id, report.ReporterPhone.MustGet(), message)
if report.ReporterPhone != "" {
log.Debug().Str("report_id", report_id).Msg("contacting via phone")
p, err := text.ParsePhoneNumber(report.ReporterPhone)
if err != nil {
return nil, fmt.Errorf("parse phone: %w", err)
}
msg_id, err := text.ReportMessage(ctx, txn, int32(user.ID), int32(report.ID), *p, message)
if err != nil {
return nil, fmt.Errorf("send text: %w", err)
}
txn.Commit(ctx)
log.Debug().Int32("msg_id", *msg_id).Msg("Created text.ReportMessage")
return msg_id, nil
} else if report.ReporterEmail.GetOr("") != "" {
msg_id, err := email.ReportMessage(ctx, int32(user.ID), report_id, report.ReporterEmail.MustGet(), message)
} else if report.ReporterEmail != "" {
msg_id, err := email.ReportMessage(ctx, int32(user.ID), report_id, report.ReporterEmail, message)
if err != nil {
return nil, fmt.Errorf("send email: %w", err)
}
txn.Commit(ctx)
return msg_id, nil
} else {
log.Debug().Str("report_id", report_id).Msg("contacting via email")
return nil, errors.New("no contact methods available")
}
}
func PublicReportReporterUpdated(ctx context.Context, org_id int32, report_id string, tablename string) {
resource := resourceTypeFromTablename(tablename)
event.Updated(resource, org_id, report_id)
event.Updated(event.TypeRMOReport, org_id, report_id)
}
func resourceTypeFromTablename(tablename string) event.ResourceType {
switch tablename {
case "nuisance":
return event.TypeRMONuisance
case "water":
return event.TypeRMOWater
default:
return event.TypeUnknown
}
}
func reportFromID(ctx context.Context, user User, report_id string) (string, *models.PublicreportReportLocation, error) {
report, err := models.PublicreportReportLocations.Query(
models.SelectWhere.PublicreportReportLocations.PublicID.EQ(report_id),
models.SelectWhere.PublicreportReportLocations.OrganizationID.EQ(user.Organization.ID()),
func reportFromID(ctx context.Context, user User, report_id string) (*models.PublicreportReport, error) {
report, err := models.PublicreportReports.Query(
models.SelectWhere.PublicreportReports.PublicID.EQ(report_id),
models.SelectWhere.PublicreportReports.OrganizationID.EQ(user.Organization.ID()),
).One(ctx, db.PGInstance.BobDB)
if err != nil {
return "", nil, err
return nil, err
}
tablename := report.TableName.MustGet()
return tablename, report, nil
return report, nil
}

70
platform/publicreport/image.go
View file

@ -28,85 +28,37 @@ LEFT JOIN publicreport.image_exif e ON i.id = e.image_id
WHERE i.id IN (1, 2, 3, 4)
GROUP BY i.id;
*/
// Get all the images that belong to the list of nuisance report IDs
func loadImagesForReportNuisance(ctx context.Context, org_id int32, report_ids []int32) (results map[int32][]types.Image, err error) {
// Get all the images that belong to the list of report IDs
func loadImagesForReport(ctx context.Context, org_id int32, report_ids []int32) (results map[int32][]types.Image, err error) {
rows, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"i.storage_uuid AS uuid",
"COALESCE(ST_X(i.location), 0) AS \"location.longitude\"",
"COALESCE(ST_Y(i.location), 0) AS \"location.latitude\"",
"ST_Distance(i.location::geography, n.location::geography) AS \"distance_from_reporter_meters\"",
"ST_Distance(i.location::geography, r.location::geography) AS \"distance_from_reporter_meters\"",
"COALESCE(MAX(e.value) FILTER (WHERE e.name = 'Make'), '') AS exif_make",
"COALESCE(MAX(e.value) FILTER (WHERE e.name = 'Model'), '') AS exif_model",
"COALESCE(MAX(e.value) FILTER (WHERE e.name = 'DateTime'), '') AS exif_datetime",
"ni.nuisance_id AS report_id",
"r.id AS report_id",
),
sm.From("publicreport.image").As("i"),
sm.LeftJoin("publicreport.image_exif").As("e").OnEQ(
psql.Quote("i", "id"),
psql.Quote("e", "image_id"),
),
sm.InnerJoin("publicreport.nuisance_image").As("ni").OnEQ(
psql.Quote("ni", "image_id"),
sm.InnerJoin("publicreport.report_image").As("ri").OnEQ(
psql.Quote("ri", "image_id"),
psql.Quote("i", "id"),
),
sm.InnerJoin("publicreport.nuisance").As("n").OnEQ(
psql.Quote("ni", "nuisance_id"),
psql.Quote("n", "id"),
sm.InnerJoin("publicreport.report").As("r").OnEQ(
psql.Quote("ri", "report_id"),
psql.Quote("r", "id"),
),
sm.Where(psql.Quote("ni", "nuisance_id").EQ(psql.Any(report_ids))),
sm.Where(psql.Quote("ri", "report_id").EQ(psql.Any(report_ids))),
sm.GroupBy(
//psql.Quote("i", "id"),
//psql.Quote("ni", "nuisance_id"),
psql.Raw("i.id, ni.nuisance_id, n.location"),
),
), scan.StructMapper[types.Image]())
if err != nil {
return nil, fmt.Errorf("get images: %w", err)
}
results = make(map[int32][]types.Image, len(report_ids))
for _, row := range rows {
r, ok := results[row.ReportID]
if !ok {
r = make([]types.Image, 0)
}
r = append(r, row)
results[row.ReportID] = r
}
return results, nil
}
// Get all the images that belong to the list of water report IDs
func loadImagesForReportWater(ctx context.Context, org_id int32, report_ids []int32) (results map[int32][]types.Image, err error) {
rows, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"i.storage_uuid AS uuid",
"COALESCE(ST_X(i.location), 0) AS \"location.longitude\"",
"COALESCE(ST_Y(i.location), 0) AS \"location.latitude\"",
"ST_Distance(i.location::geography, w.location::geography) AS \"distance_from_reporter_meters\"",
"COALESCE(MAX(e.value) FILTER (WHERE e.name = 'Make'), '') AS exif_make",
"COALESCE(MAX(e.value) FILTER (WHERE e.name = 'Model'), '') AS exif_model",
"COALESCE(MAX(e.value) FILTER (WHERE e.name = 'DateTime'), '') AS exif_datetime",
"wi.water_id AS report_id",
),
sm.From("publicreport.image").As("i"),
sm.LeftJoin("publicreport.image_exif").As("e").OnEQ(
psql.Quote("i", "id"),
psql.Quote("e", "image_id"),
),
sm.InnerJoin("publicreport.water_image").As("wi").OnEQ(
psql.Quote("wi", "image_id"),
psql.Quote("i", "id"),
),
sm.InnerJoin("publicreport.water").As("w").OnEQ(
psql.Quote("wi", "water_id"),
psql.Quote("w", "id"),
),
sm.Where(psql.Quote("wi", "water_id").EQ(psql.Any(report_ids))),
sm.GroupBy(
//psql.Quote("i", "id"),
//psql.Quote("ni", "nuisance_id"),
psql.Raw("i.id, wi.water_id, w.location"),
psql.Raw("i.id, ri.report_id, r.id, r.location"),
),
), scan.StructMapper[types.Image]())
if err != nil {

119
platform/publicreport/nuisance.go
View file

@ -3,71 +3,47 @@ package publicreport
import (
"context"
"fmt"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/sm"
//"github.com/Gleipnir-Technology/nidus-sync/config"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
//"github.com/Gleipnir-Technology/nidus-sync/db/models"
//"github.com/google/uuid"
//"github.com/rs/zerolog/log"
"github.com/stephenafamo/scan"
)
type Nuisance struct {
AdditionalInfo string `db:"additional_info" json:"additional_info"`
Address types.Address `db:"address" json:"address"`
AddressRaw string `db:"address_raw" json:"address_raw"`
Created time.Time `db:"created" json:"created"`
Duration string `db:"duration" json:"duration"`
ID int32 `db:"id" json:"-"`
Images []types.Image `db:"-" json:"images"`
IsLocationBackyard bool `db:"is_location_backyard" json:"is_location_backyard"`
IsLocationFrontyard bool `db:"is_location_frontyard" json:"is_location_frontyard"`
IsLocationGarden bool `db:"is_location_garden" json:"is_location_garden"`
IsLocationOther bool `db:"is_location_other" json:"is_location_other"`
IsLocationPool bool `db:"is_location_pool" json:"is_location_pool"`
Location types.Location `db:"location" json:"location"`
PublicID string `db:"public_id" json:"public_id"`
Reporter types.Contact `db:"reporter" json:"reporter"`
SourceContainer bool `db:"source_container" json:"source_container"`
SourceDescription string `db:"source_description" json:"source_description"`
SourceGutter bool `db:"source_gutter" json:"source_gutter"`
SourceStagnant bool `db:"source_stagnant" json:"source_stagnant"`
TODDay bool `db:"tod_day" json:"time_of_day_day"`
TODEarly bool `db:"tod_early" json:"time_of_day_early"`
TODEvening bool `db:"tod_evening" json:"time_of_day_evening"`
TODNight bool `db:"tod_night" json:"time_of_day_night"`
AdditionalInfo string `db:"additional_info" json:"additional_info"`
Duration string `db:"duration" json:"duration"`
IsLocationBackyard bool `db:"is_location_backyard" json:"is_location_backyard"`
IsLocationFrontyard bool `db:"is_location_frontyard" json:"is_location_frontyard"`
IsLocationGarden bool `db:"is_location_garden" json:"is_location_garden"`
IsLocationOther bool `db:"is_location_other" json:"is_location_other"`
IsLocationPool bool `db:"is_location_pool" json:"is_location_pool"`
ReportID int32 `db:"report_id" json:"-"`
SourceContainer bool `db:"source_container" json:"source_container"`
SourceDescription string `db:"source_description" json:"source_description"`
SourceGutter bool `db:"source_gutter" json:"source_gutter"`
SourceStagnant bool `db:"source_stagnant" json:"source_stagnant"`
TODDay bool `db:"tod_day" json:"time_of_day_day"`
TODEarly bool `db:"tod_early" json:"time_of_day_early"`
TODEvening bool `db:"tod_evening" json:"time_of_day_evening"`
TODNight bool `db:"tod_night" json:"time_of_day_night"`
}
func NuisanceReportForOrganization(ctx context.Context, org_id int32) ([]Nuisance, error) {
reports, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
func nuisancesByReportID(ctx context.Context, report_ids []int32) (map[int32]*Nuisance, error) {
rows, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"additional_info",
"address_raw AS address_raw",
"address_country AS \"address.country\"",
"address_locality AS \"address.locality\"",
"address_number AS \"address.number\"",
"address_postal_code AS \"address.postal_code\"",
"address_region AS \"address.region\"",
"address_street AS \"address.street\"",
"created",
"duration",
"id",
"is_location_backyard",
"is_location_frontyard",
"is_location_garden",
"is_location_other",
"is_location_pool",
"ST_Y(location::geometry::geometry(point, 4326)) AS \"location.latitude\"",
"ST_X(location::geometry::geometry(point, 4326)) AS \"location.longitude\"",
"public_id",
"reporter_email AS \"reporter.email\"",
"reporter_name AS \"reporter.name\"",
"reporter_phone AS \"reporter.phone\"",
"report_id",
"source_container",
"source_description",
"source_gutter",
@ -78,43 +54,32 @@ func NuisanceReportForOrganization(ctx context.Context, org_id int32) ([]Nuisanc
"tod_night",
),
sm.From("publicreport.nuisance"),
sm.Where(psql.Quote("publicreport", "nuisance", "organization_id").EQ(psql.Arg(org_id))),
sm.Where(psql.Quote("publicreport", "nuisance", "reviewed").IsNull()),
sm.Where(psql.Quote("report_id").EQ(
psql.Any(report_ids),
)),
), scan.StructMapper[Nuisance]())
if err != nil {
return nil, fmt.Errorf("get reports: %w", err)
return nil, fmt.Errorf("query nuisance: %w", err)
}
report_ids := make([]int32, len(reports))
for i, report := range reports {
report_ids[i] = report.ID
}
images_by_id, err := loadImagesForReportNuisance(ctx, org_id, report_ids)
if err != nil {
return nil, fmt.Errorf("images for report: %w", err)
}
for i := range reports {
images, ok := images_by_id[reports[i].ID]
if ok {
reports[i].Images = images
} else {
reports[i].Images = []types.Image{}
results := make(map[int32]*Nuisance, len(rows))
for _, row := range rows {
results[row.ReportID] = &Nuisance{
AdditionalInfo: row.AdditionalInfo,
Duration: row.Duration,
IsLocationBackyard: row.IsLocationBackyard,
IsLocationFrontyard: row.IsLocationFrontyard,
IsLocationGarden: row.IsLocationGarden,
IsLocationOther: row.IsLocationOther,
IsLocationPool: row.IsLocationPool,
SourceContainer: row.SourceContainer,
SourceDescription: row.SourceDescription,
SourceGutter: row.SourceGutter,
SourceStagnant: row.SourceStagnant,
TODDay: row.TODDay,
TODEarly: row.TODEarly,
TODEvening: row.TODEvening,
TODNight: row.TODNight,
}
}
return reports, nil
}
func NuisanceReportForOrganizationCount(ctx context.Context, org_id int32) (uint, error) {
type _Row struct {
Count uint `db:"count"`
}
row, err := bob.One(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"COUNT(*) AS count",
),
sm.From("publicreport.nuisance"),
sm.Where(psql.Quote("publicreport", "nuisance", "organization_id").EQ(psql.Arg(org_id))),
), scan.StructMapper[_Row]())
if err != nil {
return 0, fmt.Errorf("query count: %w", err)
}
return row.Count, nil
return results, nil
}

106
platform/publicreport/report.go Normal file
View file

@ -0,0 +1,106 @@
package publicreport
import (
"context"
"fmt"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/sm"
//"github.com/Gleipnir-Technology/nidus-sync/config"
"github.com/Gleipnir-Technology/nidus-sync/db"
//"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
//"github.com/google/uuid"
//"github.com/rs/zerolog/log"
"github.com/stephenafamo/scan"
)
type Report struct {
Address types.Address `db:"address" json:"address"`
AddressRaw string `db:"address_raw" json:"address_raw"`
Created time.Time `db:"created" json:"created"`
ID int32 `db:"id" json:"-"`
Images []types.Image `db:"images" json:"images"`
Location types.Location `db:"location" json:"location"`
Nuisance *Nuisance `db:"nuisance" json:"nuisance"`
PublicID string `db:"public_id" json:"public_id"`
Reporter types.Contact `db:"reporter" json:"reporter"`
Status string `db:"status" json:"status"`
Water *Water `db:"water" json:"water"`
}
func ReportsForOrganization(ctx context.Context, org_id int32) ([]Report, error) {
rows, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"address_country AS \"address.country\"",
"address_locality AS \"address.locality\"",
"address_number AS \"address.number\"",
"address_postal_code AS \"address.postal_code\"",
"address_raw AS address_raw",
"address_region AS \"address.region\"",
"address_street AS \"address.street\"",
"created",
"id",
"ST_Y(location::geometry::geometry(point, 4326)) AS \"location.latitude\"",
"ST_X(location::geometry::geometry(point, 4326)) AS \"location.longitude\"",
"public_id",
"reporter_email AS \"reporter.email\"",
"reporter_name AS \"reporter.name\"",
"reporter_phone AS \"reporter.phone\"",
"status",
),
sm.From("publicreport.report"),
sm.Where(psql.Quote("publicreport", "report", "organization_id").EQ(psql.Arg(org_id))),
sm.Where(psql.Quote("publicreport", "report", "reviewed").IsNull()),
), scan.StructMapper[Report]())
if err != nil {
return nil, fmt.Errorf("get reports: %w", err)
}
report_ids := make([]int32, len(rows))
for i, row := range rows {
report_ids[i] = row.ID
}
images_by_id, err := loadImagesForReport(ctx, org_id, report_ids)
if err != nil {
return nil, fmt.Errorf("images for report: %w", err)
}
nuisances_by_report_id, err := nuisancesByReportID(ctx, report_ids)
if err != nil {
return nil, fmt.Errorf("nuisances: %w", err)
}
waters_by_report_id, err := watersByReportID(ctx, report_ids)
if err != nil {
return nil, fmt.Errorf("waters: %w", err)
}
for _, row := range rows {
images, ok := images_by_id[row.ID]
if ok {
row.Images = images
} else {
row.Images = []types.Image{}
}
row.Nuisance = nuisances_by_report_id[row.ID]
row.Water = waters_by_report_id[row.ID]
}
return rows, nil
}
func ReportsForOrganizationCount(ctx context.Context, org_id int32) (uint, error) {
type _Row struct {
Count uint `db:"count"`
}
row, err := bob.One(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"COUNT(*) AS count",
),
sm.From("publicreport.report"),
sm.Where(psql.Quote("publicreport", "report", "organization_id").EQ(psql.Arg(org_id))),
), scan.StructMapper[_Row]())
if err != nil {
return 0, fmt.Errorf("query count: %w", err)
}
return row.Count, nil
}

119
platform/publicreport/water.go
View file

@ -3,7 +3,6 @@ package publicreport
import (
"context"
"fmt"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
@ -18,34 +17,25 @@ import (
)
type Water struct {
AccessComments string `db:"access_comments" json:"access_comments"`
AccessGate bool `db:"access_gate" json:"access_gate"`
AccessFence bool `db:"access_fence" json:"access_fence"`
AccessLocked bool `db:"access_locked" json:"access_locked"`
AccessDog bool `db:"access_dog" json:"access_dog"`
AccessOther bool `db:"access_other" json:"access_other"`
Address types.Address `db:"address" json:"address"`
AddressRaw string `db:"address_raw" json:"address_raw"`
Comments string `db:"comments" json:"comments"`
Created time.Time `db:"created" json:"created"`
HasAdult bool `db:"has_adult" json:"has_adult"`
HasBackyardPermission bool `db:"has_backyard_permission" json:"has_backyard_permission"`
HasLarvae bool `db:"has_larvae" json:"has_larvae"`
HasPupae bool `db:"has_pupae" json:"has_pupae"`
ID int32 `db:"id" json:"-"`
Images []types.Image `db:"-" json:"images"`
IsReporterConfidential bool `db:"is_reporter_confidential" json:"is_reporter_confidential"`
IsReporterOwner bool `db:"is_reporter_owner" json:"is_reporter_owner"`
Location types.Location `db:"location" json:"location"`
Owner types.Contact `db:"owner" json:"owner"`
PublicID string `db:"public_id" json:"public_id"`
Reporter types.Contact `db:"reporter" json:"reporter"`
ReporterContactConsent *bool `db:"reporter_contact_consent" json:"reporter_contact_consent"`
Status string `db:"status" json:"status"`
AccessComments string `db:"access_comments" json:"access_comments"`
AccessGate bool `db:"access_gate" json:"access_gate"`
AccessFence bool `db:"access_fence" json:"access_fence"`
AccessLocked bool `db:"access_locked" json:"access_locked"`
AccessDog bool `db:"access_dog" json:"access_dog"`
AccessOther bool `db:"access_other" json:"access_other"`
Comments string `db:"comments" json:"comments"`
HasAdult bool `db:"has_adult" json:"has_adult"`
HasBackyardPermission bool `db:"has_backyard_permission" json:"has_backyard_permission"`
HasLarvae bool `db:"has_larvae" json:"has_larvae"`
HasPupae bool `db:"has_pupae" json:"has_pupae"`
IsReporterConfidential bool `db:"is_reporter_confidential" json:"is_reporter_confidential"`
IsReporterOwner bool `db:"is_reporter_owner" json:"is_reporter_owner"`
Owner types.Contact `db:"owner" json:"owner"`
ReportID int32 `db:"report_id" json:"-"`
}
func WaterReportForOrganization(ctx context.Context, org_id int32) ([]Water, error) {
reports, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
func watersByReportID(ctx context.Context, report_ids []int32) (map[int32]*Water, error) {
rows, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"access_comments",
"access_gate",
@ -53,72 +43,45 @@ func WaterReportForOrganization(ctx context.Context, org_id int32) ([]Water, err
"access_locked",
"access_dog",
"access_other",
"access_gate AS address_raw",
"address_country AS \"address.country\"",
"address_locality AS \"address.locality\"",
"address_number AS \"address.number\"",
"address_postal_code AS \"address.postal_code\"",
"address_region AS \"address.region\"",
"address_street AS \"address.street\"",
"comments",
"created",
"has_adult",
"has_backyard_permission",
"has_larvae",
"has_pupae",
"id",
"is_reporter_confidential",
"is_reporter_owner",
"ST_Y(location::geometry::geometry(point, 4326)) AS \"location.latitude\"",
"ST_X(location::geometry::geometry(point, 4326)) AS \"location.longitude\"",
"owner_email AS \"owner.email\"",
"owner_name AS \"owner.name\"",
"owner_phone AS \"owner.phone\"",
"public_id",
"reporter_email AS \"reporter.email\"",
"reporter_name AS \"reporter.name\"",
"reporter_phone AS \"reporter.phone\"",
"reporter_contact_consent",
"status",
"report_id",
),
sm.From("publicreport.water"),
sm.Where(psql.Quote("publicreport", "water", "organization_id").EQ(psql.Arg(org_id))),
sm.Where(psql.Quote("publicreport", "water", "reviewed").IsNull()),
sm.Where(psql.Quote("report_id").EQ(
psql.Any(report_ids),
)),
), scan.StructMapper[Water]())
if err != nil {
return nil, fmt.Errorf("get reports: %w", err)
return nil, fmt.Errorf("query water: %w", err)
}
report_ids := make([]int32, len(reports))
for i, report := range reports {
report_ids[i] = report.ID
}
images_by_id, err := loadImagesForReportWater(ctx, org_id, report_ids)
if err != nil {
return nil, fmt.Errorf("images for report: %w", err)
}
for i := range reports {
images, ok := images_by_id[reports[i].ID]
if ok {
reports[i].Images = images
} else {
reports[i].Images = []types.Image{}
results := make(map[int32]*Water, len(rows))
for _, row := range rows {
results[row.ReportID] = &Water{
AccessComments: row.AccessComments,
AccessGate: row.AccessGate,
AccessFence: row.AccessFence,
AccessLocked: row.AccessLocked,
AccessDog: row.AccessDog,
AccessOther: row.AccessOther,
Comments: row.Comments,
HasAdult: row.HasAdult,
HasBackyardPermission: row.HasBackyardPermission,
HasLarvae: row.HasLarvae,
HasPupae: row.HasPupae,
IsReporterConfidential: row.IsReporterConfidential,
IsReporterOwner: row.IsReporterOwner,
Owner: row.Owner,
ReportID: row.ReportID,
}
}
return reports, nil
}
func WaterReportForOrganizationCount(ctx context.Context, org_id int32) (uint, error) {
type _Row struct {
Count uint `db:"count"`
}
row, err := bob.One(ctx, db.PGInstance.BobDB, psql.Select(
sm.Columns(
"COUNT(*) AS count",
),
sm.From("publicreport.water"),
sm.Where(psql.Quote("publicreport", "water", "organization_id").EQ(psql.Arg(org_id))),
), scan.StructMapper[_Row]())
if err != nil {
return 0, fmt.Errorf("query count: %w", err)
}
return row.Count, nil
return results, nil
}

134
platform/report.go Normal file
View file

@ -0,0 +1,134 @@
package platform
import (
"context"
"fmt"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/event"
"github.com/Gleipnir-Technology/nidus-sync/platform/geocode"
"github.com/Gleipnir-Technology/nidus-sync/platform/report"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
"github.com/rs/zerolog/log"
)
func ReportNuisanceCreate(ctx context.Context, setter_report models.PublicreportReportSetter, setter_nuisance models.PublicreportNuisanceSetter, latlng LatLng, address Address, images []ImageUpload) (*models.PublicreportReport, error) {
return reportCreate(ctx, setter_report, latlng, address, images, func(ctx context.Context, txn bob.Executor, report_id int32) error {
setter_nuisance.ReportID = omit.From(report_id)
_, err := models.PublicreportNuisances.Insert(&setter_nuisance).One(ctx, txn)
if err != nil {
return fmt.Errorf("Failed to create nuisance database record: %w", err)
}
return nil
})
}
func ReportWaterCreate(ctx context.Context, setter_report models.PublicreportReportSetter, setter_water models.PublicreportWaterSetter, latlng LatLng, address Address, images []ImageUpload) (*models.PublicreportReport, error) {
return reportCreate(ctx, setter_report, latlng, address, images, func(ctx context.Context, txn bob.Executor, report_id int32) error {
setter_water.ReportID = omit.From(report_id)
_, err := models.PublicreportWaters.Insert(&setter_water).One(ctx, txn)
if err != nil {
return fmt.Errorf("Failed to create water database record: %w", err)
}
return nil
})
}
type funcSetReportDetail = func(context.Context, bob.Executor, int32) error
func reportCreate(ctx context.Context, setter_report models.PublicreportReportSetter, latlng LatLng, address Address, images []ImageUpload, detail_setter funcSetReportDetail) (result *models.PublicreportReport, err error) {
txn, err := db.PGInstance.BobDB.BeginTx(ctx, nil)
if err != nil {
return nil, fmt.Errorf("create txn: %w", err)
}
defer txn.Rollback(ctx)
public_id, err := report.GenerateReportID()
if err != nil {
return nil, fmt.Errorf("create public ID: %w", err)
}
setter_report.PublicID = omit.From(public_id)
// If we've got an locality value it was set by geocoding so we should save it
var a *models.Address
if address.Locality != "" && latlng.Latitude != nil && latlng.Longitude != nil {
a, err = geocode.EnsureAddress(ctx, txn, address, types.Location{
Latitude: *latlng.Latitude,
Longitude: *latlng.Longitude,
})
if err != nil {
return nil, fmt.Errorf("Failed to ensure address: %w", err)
}
}
saved_images, err := saveImageUploads(ctx, txn, images)
if err != nil {
return nil, fmt.Errorf("Failed to save image uploads: %w", err)
}
var organization_id *int32
organization_id, err = MatchDistrict(ctx, latlng.Longitude, latlng.Latitude, images)
if err != nil {
log.Warn().Err(err).Msg("Failed to match district")
}
if a != nil {
setter_report.AddressID = omitnull.From(a.ID)
}
if organization_id != nil {
setter_report.OrganizationID = omit.FromPtr(organization_id)
}
result, err = models.PublicreportReports.Insert(&setter_report).One(ctx, txn)
if err != nil {
return nil, fmt.Errorf("Failed to create report database record: %w", err)
}
if latlng.Latitude != nil && latlng.Longitude != nil {
h3cell, _ := latlng.H3Cell()
geom_query, _ := latlng.GeometryQuery()
_, err = psql.Update(
um.Table("publicreport.report"),
um.SetCol("h3cell").ToArg(h3cell),
um.SetCol("location").To(geom_query),
um.Where(psql.Quote("id").EQ(psql.Arg(result.ID))),
).Exec(ctx, txn)
if err != nil {
return nil, fmt.Errorf("Failed to insert publicreport.report geospatial", err)
}
}
log.Info().Str("public_id", public_id).Int32("id", result.ID).Msg("Created base report")
if len(saved_images) > 0 {
setters := make([]*models.PublicreportReportImageSetter, 0)
for _, image := range saved_images {
setters = append(setters, &models.PublicreportReportImageSetter{
ImageID: omit.From(int32(image.ID)),
ReportID: omit.From(int32(result.ID)),
})
}
_, err = models.PublicreportReportImages.Insert(bob.ToMods(setters...)).Exec(ctx, txn)
if err != nil {
return nil, fmt.Errorf("Failed to save reference to images: %w", err)
}
log.Info().Int("len", len(images)).Msg("saved uploaded images")
}
err = detail_setter(ctx, txn, result.ID)
if err != nil {
return nil, fmt.Errorf("detail setter: %w", err)
}
txn.Commit(ctx)
if organization_id != nil {
event.Created(
event.TypeRMONuisance,
*organization_id,
result.PublicID,
)
}
return result, nil
}

144
platform/report/notification.go
View file

@ -5,35 +5,28 @@ import (
"crypto/rand"
"fmt"
"math/big"
"strconv"
"strings"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/db/sql"
"github.com/Gleipnir-Technology/nidus-sync/platform/email"
"github.com/Gleipnir-Technology/nidus-sync/platform/text"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
"github.com/rs/zerolog/log"
//"github.com/stephenafamo/scan"
)
func DistrictForReport(ctx context.Context, report_id string) (*models.Organization, error) {
some_report, err := findSomeReport(ctx, report_id)
report, err := reportByPublicID(ctx, db.PGInstance.BobDB, report_id)
if err != nil {
return nil, fmt.Errorf("Failed to find report %s: %w", report_id, err)
}
org_id := some_report.districtID(ctx)
if org_id == nil {
return nil, nil
}
result, e := models.FindOrganization(ctx, db.PGInstance.BobDB, *org_id)
result, e := models.FindOrganization(ctx, db.PGInstance.BobDB, report.OrganizationID)
if e != nil {
return nil, fmt.Errorf("Failed to load organization %d: %w", org_id, e)
return nil, fmt.Errorf("Failed to load organization %d: %w", report.OrganizationID, e)
}
return result, nil
}
@ -64,15 +57,15 @@ func GenerateReportID() (string, error) {
}
func RegisterNotificationEmail(ctx context.Context, txn bob.Executor, report_id string, destination string) *ErrorWithCode {
some_report, err := findSomeReport(ctx, report_id)
if err != nil {
return err
report, e := reportByPublicID(ctx, db.PGInstance.BobDB, report_id)
if e != nil {
return newInternalError(e, "Failed to find report")
}
e := email.EnsureInDB(ctx, destination)
e = email.EnsureInDB(ctx, destination)
if e != nil {
return newInternalError(e, "Failed to ensure phone is in DB")
}
err = some_report.addNotificationEmail(ctx, txn, destination)
err := addNotificationEmail(ctx, txn, report, destination)
if err != nil {
return err
}
@ -81,19 +74,19 @@ func RegisterNotificationEmail(ctx context.Context, txn bob.Executor, report_id
}
func RegisterNotificationPhone(ctx context.Context, txn bob.Executor, report_id string, phone types.E164) *ErrorWithCode {
some_report, err := findSomeReport(ctx, report_id)
if err != nil {
return err
report, e := reportByPublicID(ctx, db.PGInstance.BobDB, report_id)
if e != nil {
return newInternalError(e, "Failed to find report")
}
e := text.EnsureInDB(ctx, db.PGInstance.BobDB, phone)
e = text.EnsureInDB(ctx, db.PGInstance.BobDB, phone)
if e != nil {
return newInternalError(e, "Failed to ensure phone is in DB")
}
err = some_report.addNotificationPhone(ctx, txn, phone)
err := addNotificationPhone(ctx, txn, report, phone)
if err != nil {
return err
}
text.ReportSubscriptionConfirmationText(ctx, phone, report_id)
text.ReportSubscriptionConfirmationText(ctx, db.PGInstance.BobDB, phone, report.PublicID)
return nil
}
@ -136,64 +129,95 @@ func RegisterSubscriptionPhone(ctx context.Context, txn bob.Executor, phone type
}
func SaveReporter(ctx context.Context, txn bob.Executor, report_id string, name string, email string, phone *types.E164, has_consent bool) *ErrorWithCode {
some_report, err := findSomeReport(ctx, report_id)
if err != nil {
return err
report, e := reportByPublicID(ctx, db.PGInstance.BobDB, report_id)
if e != nil {
return newInternalError(e, "Failed to find report")
}
if name != "" {
err = some_report.updateReporterName(ctx, txn, name)
err := updateReporterName(ctx, txn, report, name)
if err != nil {
return err
}
}
if phone != nil {
err = some_report.updateReporterPhone(ctx, txn, *phone)
err := updateReporterPhone(ctx, txn, report, *phone)
if err != nil {
return err
}
}
if email != "" {
err = some_report.updateReporterEmail(ctx, txn, email)
err := updateReporterEmail(ctx, txn, report, email)
if err != nil {
return err
}
}
err = some_report.updateReporterConsent(ctx, txn, has_consent)
err := updateReporterConsent(ctx, txn, report, has_consent)
if err != nil {
return err
}
return nil
}
func findSomeReport(ctx context.Context, report_id string) (result SomeReport, err *ErrorWithCode) {
rows, e := sql.PublicreportIDTable(report_id).All(ctx, db.PGInstance.BobDB)
if e != nil {
log.Error().Err(e).Str("report_id", report_id).Msg("failed to query report ID table")
return result, newErrorWithCode("internal-error", "Failed to query report ID table: %w", e)
}
switch len(rows) {
case 0:
return result, newErrorWithCode("invalid-report-id", "No reports match the provided ID")
case 1:
break
default:
log.Error().Err(e).Str("report_id", report_id).Msg("More than one report with the provided ID, which shouldn't happen")
return result, newErrorWithCode("internal-error", "More than one report with the provided ID, which shouldn't happen")
}
row := rows[0]
report_id_str := row.ReportIds[0]
t, e := strconv.ParseInt(report_id_str, 10, 32)
if e != nil {
log.Error().Err(e).Str("report_id_str", report_id_str).Msg("Unable to parse integer reponse from database")
return result, newErrorWithCode("internal-error", "Unable to parse integer response from database")
}
switch row.FoundInTables[0] {
case "nuisance":
return newNuisance(ctx, report_id, int32(t))
case "water":
return newWater(ctx, report_id, int32(t))
default:
log.Error().Err(e).Str("table_name", row.FoundInTables[0]).Msg("Unrecognized table")
return Nuisance{}, newErrorWithCode("internal-error", fmt.Sprintf("Unrecognized table '%s'", row.FoundInTables[0]))
}
func reportByPublicID(ctx context.Context, txn bob.Executor, public_id string) (*models.PublicreportReport, error) {
return models.PublicreportReports.Query(
models.SelectWhere.PublicreportReports.PublicID.EQ(public_id),
).One(ctx, txn)
}
func addNotificationEmail(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, email string) *ErrorWithCode {
setter := models.PublicreportNotifyEmailSetter{
Created: omit.From(time.Now()),
Deleted: omitnull.FromPtr[time.Time](nil),
EmailAddress: omit.From(email),
ReportID: omit.From(report.ID),
}
_, err := models.PublicreportNotifyEmails.Insert(&setter).Exec(ctx, txn)
if err != nil {
return newInternalError(err, "Failed to save new notification email row")
}
return nil
}
func addNotificationPhone(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, phone types.E164) *ErrorWithCode {
var err error
setter := models.PublicreportNotifyPhoneSetter{
Created: omit.From(time.Now()),
Deleted: omitnull.FromPtr[time.Time](nil),
PhoneE164: omit.From(phone.PhoneString()),
ReportID: omit.From(report.ID),
}
_, err = models.PublicreportNotifyPhones.Insert(&setter).Exec(ctx, txn)
if err != nil {
return newInternalError(err, "Failed to save new notification phone row")
}
return nil
}
func updateReporterConsent(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, has_consent bool) *ErrorWithCode {
return updateReportCol(ctx, txn, report, &models.PublicreportReportSetter{
ReporterContactConsent: omitnull.From(has_consent),
})
}
func updateReporterEmail(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, email string) *ErrorWithCode {
return updateReportCol(ctx, txn, report, &models.PublicreportReportSetter{
ReporterEmail: omit.From(email),
})
}
func updateReporterName(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, name string) *ErrorWithCode {
return updateReportCol(ctx, txn, report, &models.PublicreportReportSetter{
ReporterName: omit.From(name),
})
}
func updateReportCol(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, setter *models.PublicreportReportSetter) *ErrorWithCode {
err := report.Update(ctx, txn, setter)
if err != nil {
log.Error().Err(err).Str("public_id", report.PublicID).Int32("report_id", report.ID).Msg("Failed to update report")
return newInternalError(err, "Failed to update nuisance report in the database")
}
return nil
}
func updateReporterPhone(ctx context.Context, txn bob.Executor, report *models.PublicreportReport, phone types.E164) *ErrorWithCode {
err := report.Update(ctx, txn, &models.PublicreportReportSetter{
ReporterPhone: omit.From(phone.PhoneString()),
})
if err != nil {
return newInternalError(err, "Failed to update report: %w", err)
}
return nil
}

133
platform/report/report_nuisance.go
View file

@ -1,133 +0,0 @@
package report
import (
"context"
//"crypto/rand"
//"fmt"
//"math/big"
//"strconv"
//"strings"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/sm"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
//"github.com/Gleipnir-Technology/nidus-sync/background"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
//"github.com/Gleipnir-Technology/nidus-sync/db/sql"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/rs/zerolog/log"
"github.com/stephenafamo/scan"
)
type Nuisance struct {
id int32
publicReportID string
row *models.PublicreportNuisance
}
func (sr Nuisance) PublicReportID() string {
return sr.publicReportID
}
func (sr Nuisance) addNotificationEmail(ctx context.Context, txn bob.Executor, email string) *ErrorWithCode {
setter := models.PublicreportNotifyEmailNuisanceSetter{
Created: omit.From(time.Now()),
Deleted: omitnull.FromPtr[time.Time](nil),
NuisanceID: omit.From(sr.id),
EmailAddress: omit.From(email),
}
_, err := models.PublicreportNotifyEmailNuisances.Insert(&setter).Exec(ctx, txn)
if err != nil {
return newInternalError(err, "Failed to save new notification email row")
}
return nil
}
func (sr Nuisance) addNotificationPhone(ctx context.Context, txn bob.Executor, phone types.E164) *ErrorWithCode {
var err error
setter := models.PublicreportNotifyPhoneNuisanceSetter{
Created: omit.From(time.Now()),
Deleted: omitnull.FromPtr[time.Time](nil),
NuisanceID: omit.From(sr.id),
PhoneE164: omit.From(phone.PhoneString()),
}
_, err = models.PublicreportNotifyPhoneNuisances.Insert(&setter).Exec(ctx, txn)
if err != nil {
return newInternalError(err, "Failed to save new notification phone row")
}
return nil
}
func (sr Nuisance) districtID(ctx context.Context) *int32 {
type _Row struct {
OrganizationID *int32
}
row, err := bob.One(ctx, db.PGInstance.BobDB, psql.Select(
sm.From("publicreport.nuisance"),
sm.Columns("organization_id"),
sm.Where(psql.Quote("public_id").EQ(psql.Arg(sr.publicReportID))),
), scan.StructMapper[_Row]())
if err != nil {
log.Warn().Err(err).Msg("Failed to query for organization_id")
return nil
}
return row.OrganizationID
}
func (sr Nuisance) reportID() int32 {
return sr.id
}
func (sr Nuisance) updateReporterConsent(ctx context.Context, txn bob.Executor, has_consent bool) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportNuisanceSetter{
ReporterContactConsent: omitnull.From(has_consent),
})
}
func (sr Nuisance) updateReportCol(ctx context.Context, txn bob.Executor, setter *models.PublicreportNuisanceSetter) *ErrorWithCode {
err := sr.row.Update(ctx, txn, setter)
if err != nil {
log.Error().Err(err).Str("public_id", sr.publicReportID).Int32("report_id", sr.id).Msg("Failed to update report")
return newInternalError(err, "Failed to update nuisance report in the database")
}
return nil
}
func (sr Nuisance) updateReporterEmail(ctx context.Context, txn bob.Executor, email string) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportNuisanceSetter{
ReporterEmail: omitnull.From(email),
})
}
func (sr Nuisance) updateReporterName(ctx context.Context, txn bob.Executor, name string) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportNuisanceSetter{
ReporterName: omitnull.From(name),
})
}
func (sr Nuisance) updateReporterPhone(ctx context.Context, txn bob.Executor, phone types.E164) *ErrorWithCode {
result, err := psql.Update(
um.Table("publicreport.nuisance"),
um.SetCol("reporter_phone").ToArg(phone.PhoneString()),
um.Where(psql.Quote("public_id").EQ(psql.Arg(sr.publicReportID))),
).Exec(ctx, txn)
if err != nil {
return newInternalError(err, "Failed to update report: %w", err)
}
rowcount, err := result.RowsAffected()
if err != nil {
return newInternalError(err, "Failed to get rows affected: %w", err)
}
if rowcount != 1 {
log.Warn().Str("public_report_id", sr.publicReportID).Msg("updated more than one row, which is a programmer error")
}
return nil
}
func newNuisance(ctx context.Context, public_id string, report_id int32) (Nuisance, *ErrorWithCode) {
row, err := models.FindPublicreportNuisance(ctx, db.PGInstance.BobDB, report_id)
if err != nil {
return Nuisance{}, newInternalError(err, "Failed to find nuisance report %d: %w", public_id, err)
}
return Nuisance{
id: report_id,
publicReportID: public_id,
row: row,
}, nil
}

119
platform/report/report_water.go
View file

@ -1,119 +0,0 @@
package report
import (
"context"
//"crypto/rand"
//"fmt"
//"math/big"
//"strconv"
//"strings"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/sm"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
"github.com/rs/zerolog/log"
"github.com/stephenafamo/scan"
)
type Water struct {
id int32
publicReportID string
row *models.PublicreportWater
}
func (sr Water) PublicReportID() string {
return sr.publicReportID
}
func (sr Water) addNotificationEmail(ctx context.Context, txn bob.Executor, email string) *ErrorWithCode {
setter := models.PublicreportNotifyEmailWaterSetter{
Created: omit.From(time.Now()),
Deleted: omitnull.FromPtr[time.Time](nil),
EmailAddress: omit.From(email),
WaterID: omit.From(sr.id),
}
_, err := models.PublicreportNotifyEmailWaters.Insert(&setter).Exec(ctx, txn)
if err != nil {
log.Error().Err(err).Msg("Failed to save new notification email row")
return newInternalError(err, "Failed to save new notification email row")
}
return nil
}
func (sr Water) addNotificationPhone(ctx context.Context, txn bob.Executor, phone types.E164) *ErrorWithCode {
setter := models.PublicreportNotifyPhoneWaterSetter{
Created: omit.From(time.Now()),
Deleted: omitnull.FromPtr[time.Time](nil),
PhoneE164: omit.From(phone.PhoneString()),
WaterID: omit.From(sr.id),
}
_, err := models.PublicreportNotifyPhoneWaters.Insert(&setter).Exec(ctx, txn)
if err != nil {
log.Error().Err(err).Msg("Failed to save new notification phone row")
return newInternalError(err, "Failed to save new notification phone row")
}
return nil
}
func (sr Water) districtID(ctx context.Context) *int32 {
type _Row struct {
OrganizationID *int32
}
row, err := bob.One(ctx, db.PGInstance.BobDB, psql.Select(
sm.From("publicreport.water"),
sm.Columns("organization_id"),
sm.Where(psql.Quote("public_id").EQ(psql.Arg(sr.publicReportID))),
), scan.StructMapper[_Row]())
if err != nil {
log.Warn().Err(err).Msg("Failed to query for organization_id")
return nil
}
return row.OrganizationID
}
func (sr Water) reportID() int32 {
return sr.id
}
func (sr Water) updateReporterConsent(ctx context.Context, txn bob.Executor, has_consent bool) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportWaterSetter{
ReporterContactConsent: omitnull.From(has_consent),
})
}
func (sr Water) updateReporterEmail(ctx context.Context, txn bob.Executor, email string) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportWaterSetter{
ReporterEmail: omit.From(email),
})
}
func (sr Water) updateReporterName(ctx context.Context, txn bob.Executor, name string) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportWaterSetter{
ReporterName: omit.From(name),
})
}
func (sr Water) updateReportCol(ctx context.Context, txn bob.Executor, setter *models.PublicreportWaterSetter) *ErrorWithCode {
err := sr.row.Update(ctx, txn, setter)
if err != nil {
log.Error().Err(err).Str("public_id", sr.publicReportID).Int32("report_id", sr.id).Msg("Failed to update report")
return newInternalError(err, "Failed to update water report in the database")
}
return nil
}
func (sr Water) updateReporterPhone(ctx context.Context, txn bob.Executor, phone types.E164) *ErrorWithCode {
return sr.updateReportCol(ctx, txn, &models.PublicreportWaterSetter{
ReporterPhone: omit.From(phone.PhoneString()),
})
}
func newWater(ctx context.Context, public_id string, report_id int32) (Water, *ErrorWithCode) {
row, err := models.FindPublicreportWater(ctx, db.PGInstance.BobDB, report_id)
if err != nil {
log.Error().Err(err).Msg("Failed to find water report")
return Water{}, newInternalError(err, "Failed to find water report %d: %w", public_id, err)
}
return Water{
id: report_id,
publicReportID: public_id,
row: row,
}, nil
}

14
platform/start.go
View file

@ -60,6 +60,11 @@ func StartAll(ctx context.Context) error {
defer waitGroup.Done()
refreshFieldseekerData(ctx, newOAuthTokenChannel)
}()
waitGroup.Add(1)
go func() {
defer waitGroup.Done()
listenForJobs(ctx)
}()
err = addWaitingJobs(ctx)
if err != nil {
@ -109,8 +114,10 @@ func handleJob(ctx context.Context, txn bob.Executor, job *models.Job) error {
return handleJobLabelStudioAudioCreate(ctx, txn, job.RowID)
case enums.JobtypeEmailSend:
return email.Job(ctx, txn, job.RowID)
case enums.JobtypeTextRespond:
return text.JobRespond(ctx, txn, job.RowID)
case enums.JobtypeTextSend:
return text.Job(ctx, txn, job.RowID)
return text.JobSend(ctx, txn, job.RowID)
default:
return fmt.Errorf("No handler for job type %s", string(job.Type))
}
@ -151,6 +158,7 @@ func listenAndDoOneJob(ctx context.Context) error {
}
for {
log.Debug().Msg("wait for notification")
notification, err := conn.Conn().WaitForNotification(ctx)
if err != nil {
//if !pgconn.Timeout(err) {
@ -161,12 +169,14 @@ func listenAndDoOneJob(ctx context.Context) error {
if err != nil {
return fmt.Errorf("failed to parse int from payload '%s': %w", notification.Payload, err)
}
log.Debug().Int("job_id", job_id).Msg("got notification for job")
c := bobpgx.NewConn(conn.Conn())
job, err := models.FindJob(ctx, c, int32(job_id))
if err != nil {
return fmt.Errorf("Failed to find job %d: %w", job_id, err)
}
sublog := log.With().Int32("job", job.ID).Int32("row_id", job.RowID).Str("type", string(job.Type)).Logger()
//tx, err := c.BeginTx(ctx, pgx.TxOptions{})
tx, err := conn.BeginTx(ctx, pgx.TxOptions{})
@ -176,7 +186,6 @@ func listenAndDoOneJob(ctx context.Context) error {
ctx, cancel := context.WithCancel(ctx)
txn := bobpgx.NewTx(tx, cancel)
sublog := log.With().Int32("job", job.ID).Int32("row_id", job.RowID).Logger()
err = handleJob(ctx, txn, job)
if err != nil {
sublog.Error().Err(err).Msg("failed to handle job")
@ -190,5 +199,6 @@ func listenAndDoOneJob(ctx context.Context) error {
return fmt.Errorf("delete job: %w", err)
}
txn.Commit(ctx)
sublog.Debug().Msg("job complete")
}
}

36
platform/text/job.go
View file

@ -5,21 +5,35 @@ import (
"fmt"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/nidus-sync/config"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/rs/zerolog/log"
)
func Job(ctx context.Context, txn bob.Executor, text_id int32) error {
return sendTextComplete(ctx, txn, text_id)
func JobRespond(ctx context.Context, txn bob.Executor, log_id int32) error {
return respondText(ctx, txn, log_id)
}
func ReportSubscriptionConfirmationText(ctx context.Context, destination types.E164, report_id string) error {
content := fmt.Sprintf("Thanks for submitting mosquito report %s. Text for any questions. We'll send you updates as we get them.", report_id)
origin := enums.CommsTextoriginWebsiteAction
err := sendTextBegin(ctx, *types.NewE164(&config.PhoneNumberReport), destination, content, origin, true, true)
func JobSend(ctx context.Context, txn bob.Executor, job_id int32) error {
job, err := models.FindCommsTextJob(ctx, txn, job_id)
if err != nil {
return fmt.Errorf("Failed to send initial confirmation: %w", err)
return fmt.Errorf("find text: %w", err)
}
return err
log.Debug().Int32("job.id", job.ID).Msg("completing text job")
return sendTextComplete(ctx, txn, job)
}
func handleWaitingTextJobs(ctx context.Context, txn bob.Executor, dst types.E164) error {
jobs, err := models.CommsTextJobs.Query(
models.SelectWhere.CommsTextJobs.Destination.EQ(dst.PhoneString()),
models.SelectWhere.CommsTextJobs.Completed.IsNull(),
).All(ctx, txn)
if err != nil {
return fmt.Errorf("query jobs: %w", err)
}
for _, job := range jobs {
err = sendTextComplete(ctx, txn, job)
if err != nil {
return fmt.Errorf("send text complete: %w", err)
}
}
return nil
}

81
platform/text/llm.go
View file

@ -1,9 +1,90 @@
package text
import (
"context"
"fmt"
"github.com/rs/zerolog/log"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/Gleipnir-Technology/nidus-sync/config"
"github.com/Gleipnir-Technology/nidus-sync/db"
//"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/db/sql"
"github.com/Gleipnir-Technology/nidus-sync/llm"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
//"github.com/rs/zerolog/log"
)
func SendTextFromLLM(content string) {
log.Info().Str("content", content).Msg("Pretend I sent a message")
}
func generateNextMessage(ctx context.Context, history []llm.Message, customer_phone types.E164) (llm.Message, error) {
_handle_report_status := func() (string, error) {
return "Report: ABCD-1234-5678, District: Delta MVCD, Status: scheduled, Appointment: Wednesday 3:30pm", nil
}
_handle_contact_district := func(reason string) {
log.Warn().Str("reason", reason).Msg("Contacting district")
}
_handle_contact_supervisor := func(reason string) {
log.Warn().Str("reason", reason).Msg("Contacting supervisor")
}
return llm.GenerateNextMessage(ctx, history, _handle_report_status, _handle_contact_district, _handle_contact_supervisor)
}
func handleResetConversation(ctx context.Context, txn bob.Executor, src types.E164) error {
err := wipeLLMMemory(ctx, src)
sublog := log.With().Str("src", src.PhoneString()).Logger()
if err != nil {
return fmt.Errorf("wipe memory: %w")
}
content := "LLM memory wiped"
err = sendTextCommandResponse(ctx, txn, src, content)
if err != nil {
return fmt.Errorf("Failed to indicated memory wiped: %w", err)
}
sublog.Info().Err(err).Msg("Wiped LLM memory")
return nil
}
func loadPreviousMessagesForLLM(ctx context.Context, src types.E164) ([]llm.Message, error) {
messages, err := sql.TextsBySenders(config.PhoneNumberReportStr, src.PhoneString()).All(ctx, db.PGInstance.BobDB)
results := make([]llm.Message, 0)
if err != nil {
return results, fmt.Errorf("Failed to get message history for %s and %s: %w", config.PhoneNumberReportStr, src, err)
}
for _, m := range messages {
if m.IsVisibleToLLM {
is_from_customer := (m.Source == src.PhoneString())
results = append(results, llm.Message{
IsFromCustomer: is_from_customer,
Content: m.Content,
})
}
}
return results, nil
}
func wipeLLMMemory(ctx context.Context, src types.E164) error {
destination := config.PhoneNumberReportStr
rows, err := sql.TextsBySenders(destination, src.PhoneString()).All(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("Failed to query for texts: %w", err)
}
ids := make([]int32, 0)
for _, r := range rows {
ids = append(ids, r.ID)
}
_, err = models.CommsTextLogs.Update(
um.Where(
models.CommsTextLogs.Columns.ID.EQ(psql.Any(ids)),
),
um.SetCol("is_visible_to_llm").ToArg(false),
).Exec(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("Failed to update texts: %w", err)
}
return nil
}

50
platform/text/phone_number.go Normal file
View file

@ -0,0 +1,50 @@
package text
import (
"context"
"fmt"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/im"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/rs/zerolog/log"
)
func EnsureInDB(ctx context.Context, txn bob.Executor, dst types.E164) (err error) {
return ensureInDB(ctx, txn, dst.PhoneString())
}
func ensureInDB(ctx context.Context, txn bob.Executor, destination string) (err error) {
_, err = psql.Insert(
im.Into("comms.phone", "e164", "is_subscribed", "status"),
im.Values(
psql.Arg(destination),
psql.Arg(false),
psql.Arg("unconfirmed"),
),
im.OnConflict("e164").DoNothing(),
).Exec(ctx, txn)
return err
}
func phoneStatus(ctx context.Context, src types.E164) (enums.CommsPhonestatustype, error) {
phone, err := models.FindCommsPhone(ctx, db.PGInstance.BobDB, src.PhoneString())
if err != nil {
return enums.CommsPhonestatustypeUnconfirmed, fmt.Errorf("Failed to determine if '%s' is subscribed: %w", src.PhoneString(), err)
}
return phone.Status, nil
}
func setPhoneStatus(ctx context.Context, txn bob.Executor, src types.E164, status enums.CommsPhonestatustype) error {
phone, err := models.FindCommsPhone(ctx, txn, src.PhoneString())
if err != nil {
return fmt.Errorf("Failed to determine if '%s' is subscribed: %w", src, err)
}
phone.Update(ctx, txn, &models.CommsPhoneSetter{
Status: omit.From(status),
})
log.Info().Str("src", src.PhoneString()).Str("status", string(status)).Msg("Set number subscribed")
return nil
}

42
platform/text/report-subscription.go
View file

@ -1,42 +0,0 @@
package text
import (
"context"
"fmt"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
//"github.com/rs/zerolog/log"
)
func sendReportSubscription(ctx context.Context, source, destination types.E164, content string) error {
err := EnsureInDB(ctx, db.PGInstance.BobDB, destination)
if err != nil {
return fmt.Errorf("Failed to ensure text message destination is in the DB: %w", err)
}
status, err := phoneStatus(ctx, destination)
if err != nil {
return fmt.Errorf("Failed to check if subscribed: %w", err)
}
switch status {
case enums.CommsPhonestatustypeUnconfirmed:
err = delayMessage(ctx, enums.CommsTextjobsourceRmo, destination, content, enums.CommsTextjobtypeReportConfirmation)
if err != nil {
return fmt.Errorf("Failed to delay report subscription message: %w", err)
}
err := ensureInitialText(ctx, source, destination)
if err != nil {
return fmt.Errorf("Failed to ensure initial text has been sent: %w", err)
}
return nil
case enums.CommsPhonestatustypeOkToSend:
err = sendTextBegin(ctx, source, destination, content, enums.CommsTextoriginWebsiteAction, false, true)
if err != nil {
return fmt.Errorf("Failed to send report subscription confirmation: %w", err)
}
case enums.CommsPhonestatustypeStopped:
resendInitialText(ctx, source, destination)
}
return nil
}

34
platform/text/report.go
View file

@ -2,9 +2,39 @@ package text
import (
"context"
//"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"fmt"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
//"github.com/rs/zerolog/log"
)
func ReportMessage(ctx context.Context, user_id int32, report_id, destination, message string) (*int32, error) {
// Send a message from a district to a public reporter within the context of the public report
func ReportMessage(ctx context.Context, txn bob.Executor, user_id int32, report_id int32, destination types.E164, content string) (*int32, error) {
job_id, err := sendTextBegin(ctx, txn, &user_id, &report_id, destination, content, enums.CommsTextjobtypeReportMessage)
if err != nil {
return nil, fmt.Errorf("Failed to send initial confirmation: %w", err)
}
return job_id, nil
}
// Send a message from the system to a public reporter indicating they are subscribed to updates on the report
func ReportSubscriptionConfirmationText(ctx context.Context, txn bob.Executor, destination types.E164, report_id string) error {
content := fmt.Sprintf("Thanks for submitting mosquito report %s. Text for any questions. We'll send you updates as we get them.", report_id)
_, err := sendTextBegin(ctx, txn, nil, nil, destination, content, enums.CommsTextjobtypeReportConfirmation)
if err != nil {
return fmt.Errorf("Failed to send initial confirmation: %w", err)
}
return err
}
func reportForTextRecipient(ctx context.Context, txn bob.Executor, destination types.E164) (*models.PublicreportReport, error) {
/*return models.ReportText
psql.Query(
return Addresses.Query(
sm.Where(Addresses.Columns.ID.EQ(psql.Arg(IDPK))),
).Exists(ctx, exec)
*/
return nil, nil
}

191
platform/text/send.go Normal file
View file

@ -0,0 +1,191 @@
package text
import (
"context"
"fmt"
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/nidus-sync/comms/text"
"github.com/Gleipnir-Technology/nidus-sync/config"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
//"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/background"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/rs/zerolog/log"
)
func ensureInitialText(ctx context.Context, txn bob.Executor, dst types.E164) error {
rows, err := models.CommsTextLogs.Query(
models.SelectWhere.CommsTextLogs.Destination.EQ(dst.PhoneString()),
models.SelectWhere.CommsTextLogs.IsWelcome.EQ(true),
).All(ctx, txn)
if err != nil {
return fmt.Errorf("Failed to query text logs: %w", err)
}
if len(rows) > 0 {
return nil
}
return sendInitialText(ctx, txn, dst)
}
func insertTextLog(ctx context.Context, txn bob.Executor, destination types.E164, source types.E164, origin enums.CommsTextorigin, content string, is_welcome bool, is_visible_to_llm bool) (l *models.CommsTextLog, err error) {
l, err = models.CommsTextLogs.Insert(&models.CommsTextLogSetter{
//ID:
Content: omit.From(content),
Created: omit.From(time.Now()),
Destination: omit.From(destination.PhoneString()),
IsVisibleToLLM: omit.From(is_visible_to_llm),
IsWelcome: omit.From(is_welcome),
Origin: omit.From(origin),
Source: omit.From(source.PhoneString()),
TwilioSid: omitnull.FromPtr[string](nil),
TwilioStatus: omit.From(""),
}).One(ctx, txn)
return l, err
}
func resendInitialText(ctx context.Context, txn bob.Executor, dst types.E164) error {
phone, err := models.FindCommsPhone(ctx, txn, dst.PhoneString())
if err != nil {
return fmt.Errorf("Failed to find phone %s: %w", dst, err)
}
err = phone.Update(ctx, txn, &models.CommsPhoneSetter{
Status: omit.From(enums.CommsPhonestatustypeUnconfirmed),
})
if err != nil {
return fmt.Errorf("Failed to clear subscription on phone %s: %w", dst, err)
}
return nil
}
func sendInitialText(ctx context.Context, txn bob.Executor, dst types.E164) error {
content := "Welcome to Report Mosquitoes Online. We received your request and want to confirm text updates. Reply YES to continue. Reply STOP at any time to unsubscribe"
_, err := sendTextDirect(ctx, txn, enums.CommsTextoriginWebsiteAction, dst.PhoneString(), content, false, true)
if err != nil {
return fmt.Errorf("send text: %w", err)
}
return nil
}
// Begin the process of sending the text message, but only get as far as adding it to
// the database, then let the backend finish sending.
func sendTextBegin(ctx context.Context, txn bob.Executor, user_id *int32, report_id *int32, destination types.E164, content string, type_ enums.CommsTextjobtype) (*int32, error) {
err := EnsureInDB(ctx, txn, destination)
if err != nil {
return nil, fmt.Errorf("Failed to ensure text message destination is in the DB: %w", err)
}
job, err := models.CommsTextJobs.Insert(&models.CommsTextJobSetter{
Content: omit.From(content),
CreatorID: omitnull.FromPtr(user_id),
Created: omit.From(time.Now()),
Destination: omit.From(destination.PhoneString()),
//ID:
ReportID: omitnull.FromPtr(report_id),
Source: omit.From(enums.CommsTextjobsourceRmo),
Type: omit.From(type_),
}).One(ctx, txn)
if err != nil {
return nil, fmt.Errorf("Failed to add delayed text job: %w", err)
}
err = background.NewTextSend(ctx, txn, job.ID)
if err != nil {
return nil, fmt.Errorf("new background job: %w", err)
}
return &job.ID, nil
}
func sendTextCommandResponse(ctx context.Context, txn bob.Executor, dst types.E164, content string) error {
_, err := sendTextDirect(ctx, txn, enums.CommsTextoriginCommandResponse, dst.PhoneString(), content, false, false)
return err
}
func sendTextComplete(ctx context.Context, txn bob.Executor, job *models.CommsTextJob) error {
dst, err := ParsePhoneNumber(job.Destination)
if err != nil {
return fmt.Errorf("parse phone: %w", err)
}
var origin enums.CommsTextorigin
switch job.Type {
case enums.CommsTextjobtypeReportConfirmation:
origin = enums.CommsTextoriginWebsiteAction
case enums.CommsTextjobtypeReportMessage:
origin = enums.CommsTextoriginDistrict
default:
return fmt.Errorf("incomplete switch: %s", string(job.Type))
}
status, err := phoneStatus(ctx, *dst)
if err != nil {
return fmt.Errorf("Failed to check if subscribed: %w", err)
}
log.Debug().Str("phone status", string(status)).Str("destination", job.Destination).Send()
switch status {
case enums.CommsPhonestatustypeUnconfirmed:
err := ensureInitialText(ctx, txn, *dst)
if err != nil {
return fmt.Errorf("Failed to ensure initial text has been sent: %w", err)
}
return nil
case enums.CommsPhonestatustypeOkToSend:
_, err = sendTextDirect(ctx, txn, origin, dst.PhoneString(), job.Content, true, false)
if err != nil {
return fmt.Errorf("Failed to send report subscription confirmation: %w", err)
}
return nil
case enums.CommsPhonestatustypeStopped:
resendInitialText(ctx, txn, *dst)
}
text_log, err := sendTextDirect(ctx, txn, origin, job.Destination, job.Content, true, false)
if err != nil {
return fmt.Errorf("send text direct: %w", err)
}
err = job.Update(ctx, txn, &models.CommsTextJobSetter{
Completed: omitnull.From(time.Now()),
})
if err != nil {
return fmt.Errorf("update job: %w", err)
}
if job.ReportID.IsValue() {
_, err := models.ReportTexts.Insert(&models.ReportTextSetter{
CreatorID: omit.From(job.CreatorID.MustGet()),
ReportID: omit.From(job.ReportID.MustGet()),
TextLogID: omit.From(text_log.ID),
}).One(ctx, txn)
if err != nil {
return fmt.Errorf("insert report_text: %w", err)
}
}
return nil
}
// Send a text message and save the appropriate database records.
// Send immediately using the current goroutine
func sendTextDirect(ctx context.Context, txn bob.Executor, origin enums.CommsTextorigin, destination, content string, is_visible_to_llm, is_welcome bool) (*models.CommsTextLog, error) {
text_log, err := models.CommsTextLogs.Insert(&models.CommsTextLogSetter{
//ID:
Content: omit.From(content),
Created: omit.From(time.Now()),
Destination: omit.From(destination),
IsVisibleToLLM: omit.From(is_visible_to_llm),
IsWelcome: omit.From(is_welcome),
Origin: omit.From(origin),
Source: omit.From(config.PhoneNumberReportStr),
TwilioSid: omitnull.FromPtr[string](nil),
TwilioStatus: omit.From(""),
}).One(ctx, txn)
if err != nil {
return nil, fmt.Errorf("insert text log: %w", err)
}
pid, err := text.SendText(ctx, config.VoipMSNumber, destination, content)
if err != nil {
return nil, fmt.Errorf("send text: %w", err)
}
err = text_log.Update(ctx, txn, &models.CommsTextLogSetter{
TwilioSid: omitnull.From(pid),
TwilioStatus: omit.From("created"),
})
if err != nil {
return nil, fmt.Errorf("update %w", err)
}
return text_log, nil
}

372
platform/text/text.go
View file

@ -7,17 +7,12 @@ import (
"time"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/im"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/Gleipnir-Technology/nidus-sync/comms/text"
"github.com/Gleipnir-Technology/nidus-sync/config"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/enums"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/db/sql"
"github.com/Gleipnir-Technology/nidus-sync/llm"
"github.com/Gleipnir-Technology/nidus-sync/platform/background"
"github.com/Gleipnir-Technology/nidus-sync/platform/event"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
@ -25,19 +20,7 @@ import (
"github.com/rs/zerolog/log"
)
func EnsureInDB(ctx context.Context, ex bob.Executor, destination types.E164) (err error) {
_, err = psql.Insert(
im.Into("comms.phone", "e164", "is_subscribed", "status"),
im.Values(
psql.Arg(destination.PhoneString()),
psql.Arg(false),
psql.Arg("unconfirmed"),
),
im.OnConflict("e164").DoNothing(),
).Exec(ctx, ex)
return err
}
func HandleTextMessage(ctx context.Context, source string, destination string, body string) error {
func HandleTextMessage(ctx context.Context, source string, destination string, content string) error {
src, err := ParsePhoneNumber(source)
if err != nil {
return fmt.Errorf("parse source '%s': %w", source, err)
@ -46,29 +29,80 @@ func HandleTextMessage(ctx context.Context, source string, destination string, b
if err != nil {
return fmt.Errorf("parse destination '%s': %w", destination, err)
}
_, err = insertTextLog(ctx, *dst, *src, enums.CommsTextoriginCustomer, body, false, true)
txn, err := db.PGInstance.BobDB.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("insert text log: %w", err)
return fmt.Errorf("start txn: %w", err)
}
defer txn.Rollback(ctx)
status, err := phoneStatus(ctx, *src)
if err != nil {
return fmt.Errorf("Failed to get phone status")
}
body_l := strings.TrimSpace(strings.ToLower(body))
// We don't know if they're subscribed or not.
is_visible_to_llm := true
if status == enums.CommsPhonestatustypeUnconfirmed {
is_visible_to_llm = false
}
l, err := models.CommsTextLogs.Insert(&models.CommsTextLogSetter{
//ID:
Content: omit.From(content),
Created: omit.From(time.Now()),
Destination: omit.From(dst.PhoneString()),
IsVisibleToLLM: omit.From(is_visible_to_llm),
IsWelcome: omit.From(false),
Origin: omit.From(enums.CommsTextoriginCustomer),
Source: omit.From(src.PhoneString()),
TwilioSid: omitnull.FromPtr[string](nil),
TwilioStatus: omit.From(""),
}).One(ctx, txn)
if err != nil {
return fmt.Errorf("insert text log: %w", err)
}
log.Debug().Int32("id", l.ID).Msg("insert comms text log")
err = background.NewTextRespond(ctx, txn, l.ID)
if err != nil {
return fmt.Errorf("text respond: %w", err)
}
txn.Commit(ctx)
return err
}
func respondText(ctx context.Context, txn bob.Executor, log_id int32) error {
l, err := models.FindCommsTextLog(ctx, txn, log_id)
if err != nil {
return fmt.Errorf("find comms: %w", err)
}
src, err := ParsePhoneNumber(l.Source)
if err != nil {
return fmt.Errorf("parse source: %w", err)
}
status, err := phoneStatus(ctx, *src)
if err != nil {
return fmt.Errorf("Failed to get phone status")
}
body_l := strings.TrimSpace(strings.ToLower(l.Content))
// If the user isn't confirmed for sending regular texts ensure they get a reprompt
if status == enums.CommsPhonestatustypeUnconfirmed {
switch body_l {
case "yes":
setPhoneStatus(ctx, *src, enums.CommsPhonestatustypeOkToSend)
content := "Thanks, we've confirmed your phone number. You can text STOP at any time if you change your mind"
err := sendTextBegin(ctx, *dst, *src, content, enums.CommsTextoriginCommandResponse, false, false)
err = setPhoneStatus(ctx, txn, *src, enums.CommsPhonestatustypeOkToSend)
if err != nil {
log.Error().Err(err).Msg("Failed to send confirmation response")
return fmt.Errorf("set phone status: %w", err)
}
handleWaitingTextJobs(ctx, *src)
content := "Thanks, we've confirmed your phone number. You can text STOP at any time if you change your mind"
err = sendTextCommandResponse(ctx, txn, *src, content)
if err != nil {
return fmt.Errorf("send response: %w", err)
}
handleWaitingTextJobs(ctx, txn, *src)
// We don't handle 'stop' here because we allow them to say 'stop' at any time, regardless of
// phone status.
//case "stop":
default:
content := "I have to start with either 'YES' or 'STOP' first, Which do you want?"
err = sendTextBegin(ctx, *dst, *src, content, enums.CommsTextoriginReiteration, false, false)
err = sendTextCommandResponse(ctx, txn, *src, content)
if err != nil {
log.Error().Err(err).Msg("Failed to resend initial prompt.")
}
@ -78,32 +112,48 @@ func HandleTextMessage(ctx context.Context, source string, destination string, b
switch body_l {
case "stop":
content := "You have successfully been unsubscribed. You will not receive any more messages from this number. Reply START to resubscribe."
err = sendTextBegin(ctx, *dst, *src, content, enums.CommsTextoriginCommandResponse, false, false)
err = sendTextCommandResponse(ctx, txn, *src, content)
if err != nil {
log.Error().Err(err).Msg("Failed to send unsubscribe acknowledgement.")
}
setPhoneStatus(ctx, *src, enums.CommsPhonestatustypeStopped)
setPhoneStatus(ctx, txn, *src, enums.CommsPhonestatustypeStopped)
return nil
case "reset conversation":
handleResetConversation(ctx, *src, *dst)
err = handleResetConversation(ctx, txn, *src)
if err != nil {
log.Error().Err(err).Msg("Failed to wipe memory")
content := "Failed to wipe memory"
sendTextCommandResponse(ctx, txn, *src, content)
return fmt.Errorf("reset conversation: %w", err)
}
return nil
default:
}
previous_messages, err := loadPreviousMessagesForLLM(ctx, *dst, *src)
// If we've got an open public report from this phone number then we'll let the district respond
report, err := reportForTextRecipient(ctx, txn, *src)
if err != nil {
return fmt.Errorf("has open report: %w", err)
}
if report != nil {
event.Updated(event.TypeRMOReport, report.OrganizationID, report.PublicID)
}
// Otherwise let the LLM handle the response
return respondTextLLM(ctx, txn, *src)
}
func respondTextLLM(ctx context.Context, txn bob.Executor, src types.E164) error {
previous_messages, err := loadPreviousMessagesForLLM(ctx, src)
if err != nil {
return fmt.Errorf("Failed to get previous messages: %w", err)
}
log.Info().Int("len", len(previous_messages)).Msg("passing")
sublog := log.With().Str("dst", destination).Str("src", source).Logger()
next_message, err := generateNextMessage(ctx, previous_messages, *src)
next_message, err := generateNextMessage(ctx, previous_messages, src)
if err != nil {
return fmt.Errorf("Failed to generate next message: %w", err)
}
err = sendTextBegin(ctx, *dst, *src, next_message.Content, enums.CommsTextoriginLLM, false, true)
_, err = sendTextDirect(ctx, txn, enums.CommsTextoriginLLM, src.PhoneString(), next_message.Content, true, false)
if err != nil {
return fmt.Errorf("Failed to send response text: %w", err)
}
sublog.Info().Str("content", next_message.Content).Str("body", body).Str("reply", next_message.Content).Msg("Handled text message")
return nil
}
@ -157,249 +207,3 @@ func UpdateMessageStatus(twilio_sid string, status string) {
return
}
}
func delayMessage(ctx context.Context, source enums.CommsTextjobsource, destination types.E164, content string, type_ enums.CommsTextjobtype) error {
job, err := models.CommsTextJobs.Insert(&models.CommsTextJobSetter{
Content: omit.From(content),
Created: omit.From(time.Now()),
Destination: omit.From(destination.PhoneString()),
//ID:
Source: omit.From(source),
Type: omit.From(type_),
}).One(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("Failed to add delayed text job: %w", err)
}
log.Info().Int32("id", job.ID).Msg("Created delayed text job")
return nil
}
func resendInitialText(ctx context.Context, src, dst types.E164) error {
phone, err := models.FindCommsPhone(ctx, db.PGInstance.BobDB, dst.PhoneString())
if err != nil {
return fmt.Errorf("Failed to find phone %s: %w", dst, err)
}
err = phone.Update(ctx, db.PGInstance.BobDB, &models.CommsPhoneSetter{
Status: omit.From(enums.CommsPhonestatustypeUnconfirmed),
})
if err != nil {
return fmt.Errorf("Failed to clear subscription on phone %s: %w", dst, err)
}
return nil
}
func sendInitialText(ctx context.Context, src, dst types.E164) error {
content := "Welcome to Report Mosquitoes Online. We received your request and want to confirm text updates. Reply YES to continue. Reply STOP at any time to unsubscribe"
origin := enums.CommsTextoriginWebsiteAction
err := sendTextBegin(ctx, src, dst, content, origin, true, true)
if err != nil {
return fmt.Errorf("Failed to send initial confirmation: %w", err)
}
return nil
}
func ensureInitialText(ctx context.Context, src, dst types.E164) error {
//
rows, err := models.CommsTextLogs.Query(
models.SelectWhere.CommsTextLogs.Destination.EQ(dst.PhoneString()),
models.SelectWhere.CommsTextLogs.IsWelcome.EQ(true),
).All(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("Failed to query text logs: %w", err)
}
if len(rows) > 0 {
return nil
}
return sendInitialText(ctx, src, dst)
}
func generateNextMessage(ctx context.Context, history []llm.Message, customer_phone types.E164) (llm.Message, error) {
_handle_report_status := func() (string, error) {
return "Report: ABCD-1234-5678, District: Delta MVCD, Status: scheduled, Appointment: Wednesday 3:30pm", nil
}
_handle_contact_district := func(reason string) {
log.Warn().Str("reason", reason).Msg("Contacting district")
}
_handle_contact_supervisor := func(reason string) {
log.Warn().Str("reason", reason).Msg("Contacting supervisor")
}
return llm.GenerateNextMessage(ctx, history, _handle_report_status, _handle_contact_district, _handle_contact_supervisor)
}
func handleWaitingTextJobs(ctx context.Context, src types.E164) {
jobs, err := models.CommsTextJobs.Query(
models.SelectWhere.CommsTextJobs.Destination.EQ(src.PhoneString()),
models.SelectWhere.CommsTextJobs.Completed.IsNull(),
).All(ctx, db.PGInstance.BobDB)
if err != nil {
log.Error().Err(err).Msg("Failed to query for jobs")
return
}
for _, job := range jobs {
var source string
switch job.Source {
case enums.CommsTextjobsourceRmo:
source = config.PhoneNumberReportStr
//case enums.CommsTextJobsourcenidus:
//src := config.PhoneNumebrNidusStr
default:
log.Error().Str("source", job.Source.String()).Msg("Can't support background text job.")
continue
}
p, err := phonenumbers.Parse(job.Destination, "US")
if err != nil {
log.Error().Err(err).Str("dest", job.Destination).Int32("id", job.ID).Msg("Invalid destination in job")
continue
}
dst := types.NewE164(p)
p, err = phonenumbers.Parse(source, "US")
if err != nil {
log.Error().Err(err).Str("source", source).Int32("id", job.ID).Msg("Invalid source in job")
continue
}
src := types.NewE164(p)
err = sendTextBegin(ctx, *src, *dst, job.Content, enums.CommsTextoriginWebsiteAction, false, true)
if err != nil {
log.Error().Err(err).Int32("id", job.ID).Msg("Failed to send delayed text job.")
continue
}
err = job.Update(ctx, db.PGInstance.BobDB, &models.CommsTextJobSetter{
Completed: omitnull.From(time.Now()),
})
if err != nil {
log.Error().Err(err).Int32("id", job.ID).Msg("Failed to update delayed text job.")
continue
}
}
}
func handleResetConversation(ctx context.Context, src types.E164, dst types.E164) {
err := wipeLLMMemory(ctx, src, dst)
sublog := log.With().Str("src", src.PhoneString()).Str("dst", dst.PhoneString()).Logger()
if err != nil {
sublog.Error().Err(err).Msg("Failed to wipe memory")
content := "Failed to wip memory"
err = sendTextBegin(ctx, dst, src, content, enums.CommsTextoriginCommandResponse, false, false)
if err != nil {
sublog.Error().Err(err).Msg("Failed to indicated memory wipe failure.")
}
return
}
content := "LLM memory wiped"
err = sendTextBegin(ctx, dst, src, content, enums.CommsTextoriginCommandResponse, false, false)
if err != nil {
sublog.Error().Err(err).Msg("Failed to indicated memory wiped.")
return
}
sublog.Info().Err(err).Msg("Wiped LLM memory")
}
func insertTextLog(ctx context.Context, destination types.E164, source types.E164, origin enums.CommsTextorigin, content string, is_welcome bool, is_visible_to_llm bool) (l *models.CommsTextLog, err error) {
l, err = models.CommsTextLogs.Insert(&models.CommsTextLogSetter{
//ID:
Content: omit.From(content),
Created: omit.From(time.Now()),
Destination: omit.From(destination.PhoneString()),
IsVisibleToLLM: omit.From(is_visible_to_llm),
IsWelcome: omit.From(is_welcome),
Origin: omit.From(origin),
Source: omit.From(source.PhoneString()),
TwilioSid: omitnull.FromPtr[string](nil),
TwilioStatus: omit.From(""),
}).One(ctx, db.PGInstance.BobDB)
if err != nil {
log.Debug().Int32("id", l.ID).Bool("is_visible_to_llm", is_visible_to_llm).Str("message", content).Msg("inserted text log")
}
return l, err
}
func phoneStatus(ctx context.Context, src types.E164) (enums.CommsPhonestatustype, error) {
phone, err := models.FindCommsPhone(ctx, db.PGInstance.BobDB, src.PhoneString())
if err != nil {
return enums.CommsPhonestatustypeUnconfirmed, fmt.Errorf("Failed to determine if '%s' is subscribed: %w", src.PhoneString(), err)
}
return phone.Status, nil
}
func loadPreviousMessagesForLLM(ctx context.Context, dst, src types.E164) ([]llm.Message, error) {
messages, err := sql.TextsBySenders(dst.PhoneString(), src.PhoneString()).All(ctx, db.PGInstance.BobDB)
results := make([]llm.Message, 0)
if err != nil {
return results, fmt.Errorf("Failed to get message history for %s and %s: %w", dst, src, err)
}
for _, m := range messages {
if m.IsVisibleToLLM {
is_from_customer := (m.Source == src.PhoneString())
results = append(results, llm.Message{
IsFromCustomer: is_from_customer,
Content: m.Content,
})
}
}
return results, nil
}
func sendTextBegin(ctx context.Context, source types.E164, destination types.E164, message string, origin enums.CommsTextorigin, is_welcome bool, is_visible_to_llm bool) error {
err := EnsureInDB(ctx, db.PGInstance.BobDB, destination)
if err != nil {
return fmt.Errorf("Failed to ensure text message destination is in the DB: %w", err)
}
l, err := insertTextLog(ctx, destination, source, origin, message, is_welcome, is_visible_to_llm)
if err != nil {
return fmt.Errorf("Failed to insert text message in the DB: %w", err)
}
return background.NewTextSend(ctx, db.PGInstance.BobDB, l.ID)
}
func sendTextComplete(ctx context.Context, txn bob.Executor, text_id int32) error {
text_log, err := models.FindCommsTextLog(ctx, txn, text_id)
if err != nil {
return fmt.Errorf("find text: %w", err)
}
sid, err := text.SendText(ctx, text_log.Source, text_log.Destination, text_log.Content)
if err != nil {
return fmt.Errorf("Failed to send text message: %w", err)
}
err = text_log.Update(ctx, db.PGInstance.BobDB, &models.CommsTextLogSetter{
TwilioSid: omitnull.From(sid),
TwilioStatus: omit.From("created"),
})
if err != nil {
return fmt.Errorf("Failed to update text Twilio status: %w", err)
}
return nil
}
func setPhoneStatus(ctx context.Context, src types.E164, status enums.CommsPhonestatustype) error {
phone, err := models.FindCommsPhone(ctx, db.PGInstance.BobDB, src.PhoneString())
if err != nil {
return fmt.Errorf("Failed to determine if '%s' is subscribed: %w", src, err)
}
phone.Update(ctx, db.PGInstance.BobDB, &models.CommsPhoneSetter{
Status: omit.From(status),
})
log.Info().Str("src", src.PhoneString()).Str("status", string(status)).Msg("Set number subscribed")
return nil
}
func wipeLLMMemory(ctx context.Context, src types.E164, dst types.E164) error {
rows, err := sql.TextsBySenders(dst.PhoneString(), src.PhoneString()).All(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("Failed to query for texts: %w", err)
}
ids := make([]int32, 0)
for _, r := range rows {
ids = append(ids, r.ID)
}
_, err = models.CommsTextLogs.Update(
um.Where(
models.CommsTextLogs.Columns.ID.EQ(psql.Any(ids)),
),
um.SetCol("is_visible_to_llm").ToArg(false),
).Exec(ctx, db.PGInstance.BobDB)
if err != nil {
return fmt.Errorf("Failed to update texts: %w", err)
}
return nil
}

103
platform/water.go
View file

@ -1,104 +1 @@
package platform
import (
"context"
"fmt"
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/bob/dialect/psql"
"github.com/Gleipnir-Technology/bob/dialect/psql/um"
"github.com/Gleipnir-Technology/nidus-sync/db"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/event"
"github.com/Gleipnir-Technology/nidus-sync/platform/geocode"
"github.com/Gleipnir-Technology/nidus-sync/platform/report"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/aarondl/opt/omit"
"github.com/aarondl/opt/omitnull"
"github.com/rs/zerolog/log"
)
func WaterCreate(ctx context.Context, setter models.PublicreportWaterSetter, latlng LatLng, address Address, images []ImageUpload) (public_id string, err error) {
txn, err := db.PGInstance.BobDB.BeginTx(ctx, nil)
if err != nil {
return "", fmt.Errorf("Failed to create transaction: %w", err)
}
defer txn.Rollback(ctx)
public_id, err = report.GenerateReportID()
if err != nil {
return "", fmt.Errorf("Failed to create water report public ID", err)
}
setter.PublicID = omit.From(public_id)
// If we've got an locality value it was set by geocoding so we should save it
var a *models.Address
if address.Locality != "" && latlng.Latitude != nil && latlng.Longitude != nil {
a, err = geocode.EnsureAddress(ctx, txn, address, types.Location{
Latitude: *latlng.Latitude,
Longitude: *latlng.Longitude,
})
if err != nil {
return "", fmt.Errorf("Failed to ensure address: %w", err)
}
}
saved_images, err := saveImageUploads(ctx, txn, images)
if err != nil {
return "", fmt.Errorf("Failed to save image uploads", err)
}
var organization_id *int32
organization_id, err = MatchDistrict(ctx, latlng.Longitude, latlng.Latitude, images)
if err != nil {
log.Warn().Err(err).Msg("Failed to match district")
}
if a != nil {
setter.AddressID = omitnull.From(a.ID)
}
if organization_id != nil {
setter.OrganizationID = omit.FromPtr(organization_id)
}
water, err := models.PublicreportWaters.Insert(&setter).One(ctx, txn)
if err != nil {
return "", fmt.Errorf("Failed to create database record", err)
}
if latlng.Latitude != nil && latlng.Longitude != nil {
h3cell, _ := latlng.H3Cell()
geom_query, _ := latlng.GeometryQuery()
_, err = psql.Update(
um.Table("publicreport.water"),
um.SetCol("h3cell").ToArg(h3cell),
um.SetCol("location").To(geom_query),
um.Where(psql.Quote("id").EQ(psql.Arg(water.ID))),
).Exec(ctx, txn)
if err != nil {
return "", fmt.Errorf("Failed to update publicreport.water geospatial", err)
}
}
log.Info().Int32("id", water.ID).Str("public_id", water.PublicID).Msg("Created water report")
setters := make([]*models.PublicreportWaterImageSetter, 0)
for _, image := range saved_images {
setters = append(setters, &models.PublicreportWaterImageSetter{
ImageID: omit.From(int32(image.ID)),
WaterID: omit.From(int32(water.ID)),
})
}
if len(setters) > 0 {
_, err = models.PublicreportWaterImages.Insert(bob.ToMods(setters...)).Exec(ctx, txn)
if err != nil {
return "", fmt.Errorf("Failed to save upload relationships", err)
}
}
txn.Commit(ctx)
if organization_id != nil {
event.Created(
event.TypeRMOWater,
*organization_id,
water.PublicID,
)
}
return water.PublicID, nil
}