System user

backends/rapidpro/backend.go

@@ -58,11 +58,12 @@
 	dyLogWriter  *DynamoLogWriter // all logs being written to dynamo
 	writerWG     *sync.WaitGroup
 
-	db     *sqlx.DB
-	rp     *redis.Pool
-	dynamo *dynamo.Service
-	s3     *s3x.Service
-	cw     *cwatch.Service
+	db           *sqlx.DB
+	rp           *redis.Pool
+	dynamo       *dynamo.Service
+	s3           *s3x.Service
+	cw           *cwatch.Service
+	systemUserID UserID
 
 	channelsByUUID *cache.Local[courier.ChannelUUID, *Channel]
 	channelsByAddr *cache.Local[courier.ChannelAddress, *Channel]
@@ -232,6 +233,12 @@
 	b.dyLogWriter = NewDynamoLogWriter(b.dynamo, b.writerWG)
 	b.dyLogWriter.Start()
 
+	// store the system user id
+	b.systemUserID, err = getSystemUserID(ctx, b.db)
+	if err != nil {
+		return err
+	}
+
 	// register and start our spool flushers
 	courier.RegisterFlusher(path.Join(b.config.SpoolDir, "msgs"), b.flushMsgFile)
 	courier.RegisterFlusher(path.Join(b.config.SpoolDir, "statuses"), b.flushStatusFile)

backends/rapidpro/backend_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/buger/jsonparser"
 	"github.com/gomodule/redigo/redis"
+	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/nyaruka/courier"
 	"github.com/nyaruka/courier/queue"
@@ -61,27 +62,31 @@ func testConfig() *courier.Config {
 	return config
 }
 
+func (ts *BackendTestSuite) loadSQL(path string) {
+	db, err := sqlx.Open("postgres", ts.b.config.DB)
+	noError(err)
+
+	sql, err := os.ReadFile(path)
+	noError(err)
+	db.MustExec(string(sql))
+}
+
 func (ts *BackendTestSuite) SetupSuite() {
 	ctx := context.Background()
+	cfg := testConfig()
 
 	// turn off logging
 	log.SetOutput(io.Discard)
 
-	b, err := courier.NewBackend(testConfig())
+	b, err := courier.NewBackend(cfg)
 	noError(err)
-
 	ts.b = b.(*backend)
-	must(ts.b.Start())
 
-	// read our schema sql
-	sqlSchema, err := os.ReadFile("schema.sql")
-	noError(err)
-	ts.b.db.MustExec(string(sqlSchema))
+	// load our test schema and data
+	ts.loadSQL("schema.sql")
+	ts.loadSQL("testdata.sql")
 
-	// read our testdata sql
-	sql, err := os.ReadFile("testdata.sql")
-	noError(err)
-	ts.b.db.MustExec(string(sql))
+	must(ts.b.Start())
 
 	ts.b.s3.Client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String("test-attachments")})
 	ts.b.s3.Client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String("test-logs")})

backends/rapidpro/contact.go

@@ -52,6 +52,9 @@ type Contact struct {
 	CreatedOn_  time.Time `db:"created_on"`
 	ModifiedOn_ time.Time `db:"modified_on"`
 
+	CreatedBy_  UserID `db:"created_by_id"`
+	ModifiedBy_ UserID `db:"modified_by_id"`
+
 	IsNew_ bool
 }
 
@@ -60,9 +63,10 @@ func (c *Contact) UUID() courier.ContactUUID { return c.UUID_ }
 
 const sqlInsertContact = `
 INSERT INTO 
-	contacts_contact( org_id, is_active, status, uuid,  created_on,  modified_on,  name, ticket_count) 
-              VALUES(:org_id, TRUE,      'A',   :uuid, :created_on, :modified_on, :name, 0)
-RETURNING id`
+	contacts_contact(org_id, is_active, status, uuid, created_on, modified_on, created_by_id, modified_by_id, name, ticket_count) 
+              VALUES(:org_id, TRUE, 'A', :uuid, :created_on, :modified_on, :created_by_id, :modified_by_id, :name, 0)
+RETURNING id
+`
 
 // insertContact inserts the passed in contact, the id field will be populated with the result on success
 func insertContact(tx *sqlx.Tx, contact *Contact) error {
@@ -130,7 +134,9 @@ func contactForURN(ctx context.Context, b *backend, org OrgID, channel *Channel,
 	contact.OrgID_ = org
 	contact.UUID_ = courier.ContactUUID(uuids.NewV4())
 	contact.CreatedOn_ = time.Now()
+	contact.CreatedBy_ = b.systemUserID
 	contact.ModifiedOn_ = time.Now()
+	contact.ModifiedBy_ = b.systemUserID
 	contact.IsNew_ = true
 
 	// if we aren't an anonymous org, we want to look up a name if possible and set it

backends/rapidpro/schema.sql

@@ -1,3 +1,10 @@
+DROP TABLE IF EXISTS users_user CASCADE;
+CREATE TABLE users_user (
+    id serial primary key,
+    username character varying(254) NOT NULL,
+    first_name character varying(150) NOT NULL
+);
+
 DROP TABLE IF EXISTS orgs_org CASCADE;
 CREATE TABLE orgs_org (
     id serial primary key,
@@ -36,8 +43,8 @@ CREATE TABLE contacts_contact (
     uuid character varying(36) NOT NULL,
     name character varying(128),
     language character varying(3),
-    created_by_id integer,
-    modified_by_id integer,
+    created_by_id integer references users_user(id) NOT NULL,
+    modified_by_id integer references users_user(id) NOT NULL,
     org_id integer references orgs_org(id) on delete cascade
 );
 
@@ -89,7 +96,7 @@ CREATE TABLE msgs_msg (
     --broadcast_id integer REFERENCES msgs_broadcast(id) ON DELETE CASCADE,
     --flow_id integer REFERENCES flows_flow(id) ON DELETE CASCADE,
     --ticket_id integer REFERENCES tickets_ticket(id) ON DELETE CASCADE,
-    --created_by_id integer REFERENCES auth_user(id) ON DELETE CASCADE,
+    created_by_id integer REFERENCES users_user(id),
     text text NOT NULL,
     attachments character varying(255)[] NULL,
     quick_replies character varying(64)[] NULL,

backends/rapidpro/testdata.sql

@@ -1,3 +1,6 @@
+DELETE FROM users_user;
+INSERT INTO users_user("id", "username", "first_name") VALUES(1, 'system', 'System');
+
 /* Org with id 1 */
 DELETE FROM orgs_org;
 INSERT INTO orgs_org("id", "name", "language", "is_anon", "config")

backends/rapidpro/user.go

@@ -0,0 +1,20 @@
+package rapidpro
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jmoiron/sqlx"
+)
+
+type UserID int
+
+// gets the system user to use for contact audit fields
+func getSystemUserID(ctx context.Context, db *sqlx.DB) (UserID, error) {
+	var id UserID
+	err := db.GetContext(ctx, &id, "SELECT id FROM users_user WHERE username = 'system'")
+	if err != nil {
+		return 0, fmt.Errorf("error looking up system user: %w", err)
+	}
+	return id, nil
+}