No activity yet
diff --git a/llm/openai.go b/llm/openai.go
index 1712100d..1b643aac 100644
--- a/llm/openai.go
+++ b/llm/openai.go
@@ -2,6 +2,7 @@ package llm
import (
"context"
+ "errors"
"fmt"
"strings"
@@ -50,6 +51,9 @@ type QueryReportStatusInput struct {
var client *openAIClient
func (c *openAIClient) continueConversation(ctx context.Context, tools genai.OptionsTools, msg genai.Message) (Message, error) {
+ if c.client == nil {
+ return Message{}, errors.New("Client not initialized")
+ }
res, _, err := adapters.GenSyncWithToolCallLoop(ctx, c.client, genai.Messages{msg}, &tools)
if err != nil {
return Message{}, fmt.Errorf("Failed to continue conversation: %v", err)
diff --git a/platform/publicreport.go b/platform/publicreport.go
index d16055f1..5ef67781 100644
--- a/platform/publicreport.go
+++ b/platform/publicreport.go
@@ -6,16 +6,21 @@ import (
"fmt"
"time"
- "github.com/aarondl/opt/omit"
- "github.com/aarondl/opt/omitnull"
- //"github.com/Gleipnir-Technology/bob"
+ "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"
+ "github.com/aarondl/opt/omit"
+ "github.com/aarondl/opt/omitnull"
//"github.com/Gleipnir-Technology/nidus-sync/platform/background"
"github.com/Gleipnir-Technology/nidus-sync/platform/email"
"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/text"
+ "github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/rs/zerolog/log"
)
@@ -58,7 +63,7 @@ func PublicReportMessageCreate(ctx context.Context, user User, report_id, messag
return nil, fmt.Errorf("send text: %w", err)
}
txn.Commit(ctx)
- log.Debug().Int32("msg_id", *msg_id).Msg("Created text.ReportMessage")
+ //log.Debug().Int32("msg_id", *msg_id).Msg("Created text.ReportMessage")
return msg_id, nil
} else if report.ReporterEmail != "" {
msg_id, err := email.ReportMessage(ctx, int32(user.ID), report_id, report.ReporterEmail, message)
@@ -72,9 +77,134 @@ func PublicReportMessageCreate(ctx context.Context, user User, report_id, messag
return nil, errors.New("no contact methods available")
}
}
-func PublicReportReporterUpdated(ctx context.Context, org_id int32, report_id string, tablename string) {
+func PublicReportReporterUpdated(ctx context.Context, org_id int32, report_id string) {
event.Updated(event.TypeRMOReport, org_id, report_id)
}
+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)
+ }
+
+ models.PublicreportReportLogs.Insert(&models.PublicreportReportLogSetter{
+ Created: omit.From(time.Now()),
+ EmailLogID: omitnull.FromPtr[int32](nil),
+ // ID
+ ReportID: omit.From(result.ID),
+ TextLogID: omitnull.FromPtr[int32](nil),
+ Type: omit.From(enums.PublicreportReportlogtypeCreated),
+ UserID: omitnull.FromPtr[int32](nil),
+ }).One(ctx, txn)
+
+ txn.Commit(ctx)
+
+ if organization_id != nil {
+ event.Created(
+ event.TypeRMONuisance,
+ *organization_id,
+ result.PublicID,
+ )
+ }
+ return result, nil
+}
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),
diff --git a/platform/publicreport/log.go b/platform/publicreport/log.go
index de2608f6..b32457d0 100644
--- a/platform/publicreport/log.go
+++ b/platform/publicreport/log.go
@@ -32,6 +32,7 @@ func logEntriesByReportID(ctx context.Context, report_ids []int32) (map[int32][]
sm.Columns(
"l.created",
"l.id",
+ "COALESCE(t.content, '') AS message",
"l.report_id",
"l.type_",
"l.user_id",
diff --git a/platform/publicreport/report.go b/platform/publicreport/report.go
index 93e6913b..c584e24c 100644
--- a/platform/publicreport/report.go
+++ b/platform/publicreport/report.go
@@ -18,6 +18,7 @@ import (
)
type Report struct {
+ Log []LogEntry `db:"-" json:"log"`
Address types.Address `db:"address" json:"address"`
AddressRaw string `db:"address_raw" json:"address_raw"`
Created time.Time `db:"created" json:"created"`
@@ -31,7 +32,7 @@ type Report struct {
Water *Water `db:"water" json:"water"`
}
-func ReportsForOrganization(ctx context.Context, org_id int32) ([]Report, error) {
+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\"",
@@ -67,6 +68,10 @@ func ReportsForOrganization(ctx context.Context, org_id int32) ([]Report, error)
if err != nil {
return nil, fmt.Errorf("images for report: %w", err)
}
+ logs_by_report_id, err := logEntriesByReportID(ctx, report_ids)
+ if err != nil {
+ return nil, fmt.Errorf("log entries for reports: %w", err)
+ }
nuisances_by_report_id, err := nuisancesByReportID(ctx, report_ids)
if err != nil {
return nil, fmt.Errorf("nuisances: %w", err)
@@ -76,17 +81,20 @@ func ReportsForOrganization(ctx context.Context, org_id int32) ([]Report, error)
return nil, fmt.Errorf("waters: %w", err)
}
- for _, row := range rows {
+ results := make([]*Report, len(rows))
+ for i, row := range rows {
images, ok := images_by_id[row.ID]
if ok {
row.Images = images
} else {
row.Images = []types.Image{}
}
+ row.Log = logs_by_report_id[row.ID]
row.Nuisance = nuisances_by_report_id[row.ID]
row.Water = waters_by_report_id[row.ID]
+ results[i] = &row
}
- return rows, nil
+ return results, nil
}
func ReportsForOrganizationCount(ctx context.Context, org_id int32) (uint, error) {
type _Row struct {
diff --git a/platform/start.go b/platform/start.go
index 41e35b85..25946aa2 100644
--- a/platform/start.go
+++ b/platform/start.go
@@ -169,7 +169,7 @@ 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")
+ //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))
@@ -199,6 +199,6 @@ func listenAndDoOneJob(ctx context.Context) error {
return fmt.Errorf("delete job: %w", err)
}
txn.Commit(ctx)
- sublog.Debug().Msg("job complete")
+ //sublog.Debug().Msg("job complete")
}
}
diff --git a/platform/text/job.go b/platform/text/job.go
index 494641e6..347028aa 100644
--- a/platform/text/job.go
+++ b/platform/text/job.go
@@ -7,7 +7,7 @@ import (
"github.com/Gleipnir-Technology/bob"
"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
- "github.com/rs/zerolog/log"
+ //"github.com/rs/zerolog/log"
)
func JobRespond(ctx context.Context, txn bob.Executor, log_id int32) error {
@@ -18,7 +18,7 @@ func JobSend(ctx context.Context, txn bob.Executor, job_id int32) error {
if err != nil {
return fmt.Errorf("find text: %w", err)
}
- log.Debug().Int32("job.id", job.ID).Msg("completing text job")
+ //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 {
diff --git a/platform/text/report.go b/platform/text/report.go
index 5bc9dd95..2a836d6d 100644
--- a/platform/text/report.go
+++ b/platform/text/report.go
@@ -5,10 +5,14 @@ import (
"fmt"
"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/enums"
- "github.com/Gleipnir-Technology/nidus-sync/db/models"
+ //"github.com/Gleipnir-Technology/nidus-sync/db/models"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
//"github.com/rs/zerolog/log"
+ "github.com/stephenafamo/scan"
)
// Send a message from a district to a public reporter within the context of the public report
@@ -29,12 +33,35 @@ func ReportSubscriptionConfirmationText(ctx context.Context, txn bob.Executor, d
}
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
+
+type reportIDs struct {
+ ID int32 `db:"id"`
+ PublicID string `db:"public_id"`
+ OrganizationID int32 `db:"organization_id"`
+}
+
+// Get the list of reports that are still open for a particular text message recipient
+// 'still open' is not well-defined throughout the system, but for now we'll go with
+// 'not reviewed in any way'.
+func reportsForTextRecipient(ctx context.Context, txn bob.Executor, destination types.E164) ([]reportIDs, error) {
+ rows, err := bob.All(ctx, db.PGInstance.BobDB, psql.Select(
+ sm.Columns(
+ "r.id",
+ "r.public_id",
+ "r.organization_id",
+ ),
+ sm.From("comms.text_job").As("t"),
+ sm.InnerJoin("publicreport.report").As("r").OnEQ(
+ psql.Quote("t", "report_id"),
+ psql.Quote("r", "id"),
+ ),
+ sm.Where(psql.Quote("t", "report_id").IsNotNull()),
+ sm.Where(psql.Quote("t", "destination").EQ(psql.Arg(destination.PhoneString()))),
+ sm.Where(psql.Quote("r", "status").EQ(psql.Arg(enums.PublicreportReportstatustypeReported))),
+ ), scan.StructMapper[reportIDs]())
+ if err != nil {
+ return []reportIDs{}, fmt.Errorf("query reports: %w", err)
+ }
+
+ return rows, nil
}
diff --git a/platform/text/send.go b/platform/text/send.go
index 03ade096..569033fc 100644
--- a/platform/text/send.go
+++ b/platform/text/send.go
@@ -14,6 +14,7 @@ import (
"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/event"
"github.com/Gleipnir-Technology/nidus-sync/platform/types"
"github.com/rs/zerolog/log"
)
@@ -125,14 +126,11 @@ func sendTextComplete(ctx context.Context, txn bob.Executor, job *models.CommsTe
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.CommsPhonestatustypeOkToSend:
+ // allow to drop through
case enums.CommsPhonestatustypeStopped:
resendInitialText(ctx, txn, *dst)
+ return nil
}
text_log, err := sendTextDirect(ctx, txn, origin, job.Destination, job.Content, true, false)
if err != nil {
@@ -145,14 +143,33 @@ func sendTextComplete(ctx context.Context, txn bob.Executor, job *models.CommsTe
return fmt.Errorf("update job: %w", err)
}
if job.ReportID.IsValue() {
+ creator_id := job.CreatorID.MustGet()
+ report_id := job.ReportID.MustGet()
+ log.Debug().Int32("creator", creator_id).Int32("report_id", report_id).Msg("Creating report entries for text message")
_, err := models.ReportTexts.Insert(&models.ReportTextSetter{
- CreatorID: omit.From(job.CreatorID.MustGet()),
- ReportID: omit.From(job.ReportID.MustGet()),
+ CreatorID: omit.From(creator_id),
+ ReportID: omit.From(report_id),
TextLogID: omit.From(text_log.ID),
}).One(ctx, txn)
if err != nil {
return fmt.Errorf("insert report_text: %w", err)
}
+ models.PublicreportReportLogs.Insert(&models.PublicreportReportLogSetter{
+ Created: omit.From(time.Now()),
+ EmailLogID: omitnull.FromPtr[int32](nil),
+ // ID
+ ReportID: omit.From(report_id),
+ TextLogID: omitnull.From(text_log.ID),
+ Type: omit.From(enums.PublicreportReportlogtypeMessageText),
+ UserID: omitnull.From(creator_id),
+ }).One(ctx, txn)
+ report, err := models.FindPublicreportReport(ctx, txn, report_id)
+ if err != nil {
+ return fmt.Errorf("find public report: %w", err)
+ }
+ event.Updated(event.TypeRMOReport, report.OrganizationID, report.PublicID)
+ } else {
+ log.Debug().Msg("no report info on text")
}
return nil
}
diff --git a/platform/text/text.go b/platform/text/text.go
index 931925b7..cbc424dd 100644
--- a/platform/text/text.go
+++ b/platform/text/text.go
@@ -129,13 +129,26 @@ func respondText(ctx context.Context, txn bob.Executor, log_id int32) error {
return nil
}
// 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)
+ reports, err := reportsForTextRecipient(ctx, txn, *src)
if err != nil {
return fmt.Errorf("has open report: %w", err)
}
- if report != nil {
+ for _, report := range reports {
+ models.PublicreportReportLogs.Insert(&models.PublicreportReportLogSetter{
+ Created: omit.From(time.Now()),
+ EmailLogID: omitnull.FromPtr[int32](nil),
+ // ID
+ ReportID: omit.From(report.ID),
+ TextLogID: omitnull.From(log_id),
+ Type: omit.From(enums.PublicreportReportlogtypeMessageText),
+ UserID: omitnull.FromPtr[int32](nil),
+ }).One(ctx, txn)
event.Updated(event.TypeRMOReport, report.OrganizationID, report.PublicID)
}
+ // If humans are involved, wait for them.
+ if len(reports) > 0 {
+ return nil
+ }
// Otherwise let the LLM handle the response
return respondTextLLM(ctx, txn, *src)
}