syz-cluster: support multiple reporter types

Introduce a Reporter column to the SessionReport.
For finished reports, store both a MessageID instead of Link.

6 files changed, 30 insertions, 21 deletions

diff --git a/syz-cluster/pkg/api/reporter.go b/syz-cluster/pkg/api/reporter.go
index dc85db45b..9b0f55d03 100644
--- a/syz-cluster/pkg/api/reporter.go
+++ b/syz-cluster/pkg/api/reporter.go
@@ -5,6 +5,7 @@ package api
 
 import (
 	"context"
+	"net/url"
 	"strings"
 )
 
@@ -20,16 +21,19 @@ type NextReportResp struct {
 	Report *SessionReport `json:"report"`
 }
 
-func (client ReporterClient) GetNextReport(ctx context.Context) (*NextReportResp, error) {
-	return postJSON[any, NextReportResp](ctx, client.baseURL+"/reports", nil)
-}
+const EmailReporter = "email"
 
-// TODO: What to do if sending the report failed? Retry or mark as failed?
+func (client ReporterClient) GetNextReport(ctx context.Context, reporter string) (*NextReportResp, error) {
+	v := url.Values{}
+	v.Add("reporter", reporter)
+	return postJSON[any, NextReportResp](ctx, client.baseURL+"/reports?"+v.Encode(), nil)
+}
 
 type UpdateReportReq struct {
-	Link string `json:"link"`
+	MessageID string `json:"message_id"`
 }
 
+// UpdateReport may be used to remember the message ID and the link to the discussion.
 func (client ReporterClient) UpdateReport(ctx context.Context, id string, req *UpdateReportReq) error {
 	_, err := postJSON[UpdateReportReq, any](ctx, client.baseURL+"/reports/"+id+"/update", req)
 	return err
diff --git a/syz-cluster/pkg/db/entities.go b/syz-cluster/pkg/db/entities.go
index 531861800..a7f08f9ab 100644
--- a/syz-cluster/pkg/db/entities.go
+++ b/syz-cluster/pkg/db/entities.go
@@ -134,7 +134,8 @@ type SessionReport struct {
 	SessionID  string           `spanner:"SessionID"`
 	ReportedAt spanner.NullTime `spanner:"ReportedAt"`
 	Moderation bool             `spanner:"Moderation"`
-	Link       string           `spanner:"Link"`
+	MessageID  string           `spanner:"MessageID"`
+	Reporter   string           `spanner:"Reporter"`
 }
 
 func (s *SessionReport) SetReportedAt(t time.Time) {
diff --git a/syz-cluster/pkg/db/migrations/1_initialize.up.sql b/syz-cluster/pkg/db/migrations/1_initialize.up.sql
index 9935774c4..ca91b343a 100644
--- a/syz-cluster/pkg/db/migrations/1_initialize.up.sql
+++ b/syz-cluster/pkg/db/migrations/1_initialize.up.sql
@@ -104,9 +104,10 @@ CREATE TABLE SessionReports (
     SessionID STRING(36) NOT NULL, -- UUID
     ReportedAt TIMESTAMP,
     Moderation BOOL,
-    Link STRING(256),
+    MessageID STRING(256),
+    Reporter STRING(256),
     CONSTRAINT FK_SessionReports FOREIGN KEY (SessionID) REFERENCES Sessions (ID),
 ) PRIMARY KEY(ID);
 
 CREATE UNIQUE INDEX NoDupSessionReports ON SessionReports(SessionID, Moderation);
-CREATE INDEX SessionReportsByStatus ON SessionReports (Moderation, ReportedAt);
+CREATE INDEX SessionReportsByStatus ON SessionReports (Reporter, ReportedAt);
diff --git a/syz-cluster/pkg/db/report_repo.go b/syz-cluster/pkg/db/report_repo.go
index e1310d21c..4e039c4a7 100644
--- a/syz-cluster/pkg/db/report_repo.go
+++ b/syz-cluster/pkg/db/report_repo.go
@@ -33,10 +33,13 @@ func (repo *ReportRepository) Insert(ctx context.Context, rep *SessionReport) er
 	return repo.genericEntityOps.Insert(ctx, rep)
 }
 
-func (repo *ReportRepository) ListNotReported(ctx context.Context, limit int) ([]*SessionReport, error) {
+func (repo *ReportRepository) ListNotReported(ctx context.Context, reporter string,
+	limit int) ([]*SessionReport, error) {
 	stmt := spanner.Statement{
-		SQL:    "SELECT * FROM `SessionReports` WHERE `ReportedAt` IS NULL",
-		Params: map[string]interface{}{},
+		SQL: "SELECT * FROM `SessionReports` WHERE `Reporter` = @reporter AND `ReportedAt` IS NULL",
+		Params: map[string]interface{}{
+			"reporter": reporter,
+		},
 	}
 	addLimit(&stmt, limit)
 	return repo.readEntities(ctx, stmt)
diff --git a/syz-cluster/pkg/db/report_repo_test.go b/syz-cluster/pkg/db/report_repo_test.go
index ee05a2e18..bc81c37bd 100644
--- a/syz-cluster/pkg/db/report_repo_test.go
+++ b/syz-cluster/pkg/db/report_repo_test.go
@@ -12,6 +12,8 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+const dummyReporter = "abcd"
+
 func TestReportRepository(t *testing.T) {
 	client, ctx := NewTransientDB(t)
 	sessionRepo := NewSessionRepository(client)
@@ -28,13 +30,13 @@ func TestReportRepository(t *testing.T) {
 		err = sessionRepo.Insert(ctx, session)
 		assert.NoError(t, err)
 
-		report := &SessionReport{SessionID: session.ID}
+		report := &SessionReport{SessionID: session.ID, Reporter: dummyReporter}
 		err = reportRepo.Insert(ctx, report)
 		assert.NoError(t, err)
 		keys = append(keys, report.ID)
 	}
 
-	list, err := reportRepo.ListNotReported(ctx, 10)
+	list, err := reportRepo.ListNotReported(ctx, dummyReporter, 10)
 	assert.NoError(t, err)
 	assert.Len(t, list, 3)
 
@@ -45,7 +47,7 @@ func TestReportRepository(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Now one less.
-	list, err = reportRepo.ListNotReported(ctx, 10)
+	list, err = reportRepo.ListNotReported(ctx, dummyReporter, 10)
 	assert.NoError(t, err)
 	assert.Len(t, list, 2)
 }
@@ -105,7 +107,7 @@ func TestSessionsWithoutReports(t *testing.T) {
 
 	// Create a report for the first session.
 	reportRepo := NewReportRepository(client)
-	err = reportRepo.Insert(ctx, &SessionReport{SessionID: sessions[0].ID})
+	err = reportRepo.Insert(ctx, &SessionReport{SessionID: sessions[0].ID, Reporter: dummyReporter})
 	assert.NoError(t, err)
 
 	// Now only the second session must be returned.
diff --git a/syz-cluster/pkg/service/report.go b/syz-cluster/pkg/service/report.go
index 4e8bc4974..5e6b6756a 100644
--- a/syz-cluster/pkg/service/report.go
+++ b/syz-cluster/pkg/service/report.go
@@ -31,11 +31,8 @@ func NewReportService(env *app.AppEnvironment) *ReportService {
 var ErrReportNotFound = errors.New("report is not found")
 
 func (rs *ReportService) Update(ctx context.Context, id string, req *api.UpdateReportReq) error {
-	// TODO: validate the link?
 	err := rs.reportRepo.Update(ctx, id, func(rep *db.SessionReport) error {
-		if req.Link != "" {
-			rep.Link = req.Link
-		}
+		rep.MessageID = req.MessageID
 		return nil
 	})
 	if errors.Is(err, db.ErrEntityNotFound) {
@@ -72,6 +69,7 @@ func (rs *ReportService) Upstream(ctx context.Context, id string, req *api.Upstr
 	// prevent duplications.
 	err = rs.reportRepo.Insert(ctx, &db.SessionReport{
 		SessionID: rep.SessionID,
+		Reporter:  rep.Reporter,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to schedule a new report: %w", err)
@@ -79,8 +77,8 @@ func (rs *ReportService) Upstream(ctx context.Context, id string, req *api.Upstr
 	return nil
 }
 
-func (rs *ReportService) Next(ctx context.Context) (*api.NextReportResp, error) {
-	list, err := rs.reportRepo.ListNotReported(ctx, 1)
+func (rs *ReportService) Next(ctx context.Context, reporter string) (*api.NextReportResp, error) {
+	list, err := rs.reportRepo.ListNotReported(ctx, reporter, 1)
 	if err != nil {
 		return nil, err
 	} else if len(list) != 1 {