Merge pull request #113 from nyaruka/delete_msg_task

Add contact task to delete a message

core/models/msgs.go

@@ -2,6 +2,7 @@ package models
 
 import (
 	"context"
+	"database/sql"
 	"database/sql/driver"
 	"fmt"
 	"strings"
@@ -746,6 +747,33 @@ func NewMsgOut(oa *OrgAssets, c *flows.Contact, text string, atts []utils.Attach
 	return flows.NewMsgOut(urn, channelRef, text, atts, qrs, nil, flows.NilMsgTopic, locale, unsendableReason), channel
 }
 
+const sqlUpdateMsgDeletedBySender = `
+UPDATE msgs_msg
+   SET visibility = 'X', text = '', attachments = '{}'
+ WHERE id = $1 AND org_id = $2 AND direction = 'I'`
+
+func UpdateMessageDeletedBySender(ctx context.Context, db *sql.DB, orgID OrgID, msgID MsgID) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.Wrap(err, "error beginning transaction")
+	}
+
+	res, err := tx.ExecContext(ctx, sqlUpdateMsgDeletedBySender, msgID, orgID)
+	if err != nil {
+		return errors.Wrap(err, "error updating message visibility")
+	}
+
+	// if there was such a message, remove its labels too
+	if rows, _ := res.RowsAffected(); rows == 1 {
+		_, err = tx.ExecContext(ctx, `DELETE FROM msgs_msg_labels WHERE msg_id = $1`, msgID)
+		if err != nil {
+			return errors.Wrap(err, "error removing message labels")
+		}
+	}
+
+	return errors.Wrap(tx.Commit(), "error committing transaction")
+}
+
 // NilID implementations
 
 func (i *MsgID) Scan(value any) error         { return null.ScanInt(value, i) }

core/models/msgs_test.go

@@ -345,6 +345,31 @@ func TestFailMessages(t *testing.T) {
 	assertdb.Query(t, rt.DB, `SELECT status, failed_reason FROM msgs_msg WHERE id = $1`, out3.ID).Columns(map[string]any{"status": "F", "failed_reason": nil})
 }
 
+func TestUpdateMessageDeletedBySender(t *testing.T) {
+	ctx, rt := testsuite.Runtime()
+
+	defer testsuite.Reset(testsuite.ResetData)
+
+	in1 := testdata.InsertIncomingMsg(rt, testdata.Org1, testdata.TwilioChannel, testdata.Cathy, "hi", models.MsgStatusHandled)
+	in1.Label(rt, testdata.ReportingLabel, testdata.TestingLabel)
+	in2 := testdata.InsertIncomingMsg(rt, testdata.Org1, testdata.TwilioChannel, testdata.Cathy, "bye", models.MsgStatusHandled)
+	in2.Label(rt, testdata.ReportingLabel, testdata.TestingLabel)
+	out1 := testdata.InsertOutgoingMsg(rt, testdata.Org1, testdata.TwilioChannel, testdata.Cathy, "hi", nil, models.MsgStatusSent, false)
+
+	err := models.UpdateMessageDeletedBySender(ctx, rt.DB.DB, testdata.Org1.ID, in1.ID)
+	assert.NoError(t, err)
+
+	assertdb.Query(t, rt.DB, `SELECT visibility, text FROM msgs_msg WHERE id = $1`, in1.ID).Columns(map[string]any{"visibility": "X", "text": ""})
+	assertdb.Query(t, rt.DB, `SELECT count(*) FROM msgs_msg_labels WHERE msg_id = $1`, in1.ID).Returns(0)
+	assertdb.Query(t, rt.DB, `SELECT count(*) FROM msgs_msg_labels WHERE msg_id = $1`, in2.ID).Returns(2) // unchanged
+
+	// trying to delete an outgoing message is a noop
+	err = models.UpdateMessageDeletedBySender(ctx, rt.DB.DB, testdata.Org1.ID, out1.ID)
+	assert.NoError(t, err)
+
+	assertdb.Query(t, rt.DB, `SELECT visibility, text FROM msgs_msg WHERE id = $1`, out1.ID).Columns(map[string]any{"visibility": "V", "text": "hi"})
+}
+
 func TestGetMsgRepetitions(t *testing.T) {
 	_, rt := testsuite.Runtime()
 

core/tasks/handler/contact_tasks.go

@@ -37,6 +37,7 @@ const (
 	ExpirationEventType      = "expiration_event"
 	TimeoutEventType         = "timeout_event"
 	TicketClosedEventType    = "ticket_closed"
+	MsgDeletedType           = "msg_deleted"
 )
 
 // handleTimedEvent is called for timeout events
@@ -605,6 +606,11 @@ func handleTicketEvent(ctx context.Context, rt *runtime.Runtime, event *models.T
 	return nil
 }
 
+func handleMsgDeletedEvent(ctx context.Context, rt *runtime.Runtime, event *MsgDeletedEvent) error {
+	err := models.UpdateMessageDeletedBySender(ctx, rt.DB.DB, event.OrgID, event.MsgID)
+	return errors.Wrap(err, "error deleting message")
+}
+
 // handles a message as an inbox message, i.e. no flow
 func handleAsInbox(ctx context.Context, rt *runtime.Runtime, oa *models.OrgAssets, contact *flows.Contact, msg *flows.MsgIn, attachments []utils.Attachment, logUUIDs []models.ChannelLogUUID, ticket *models.Ticket) error {
 	// usually last_seen_on is updated by handling the msg_received event in the engine sprint, but since this is an inbox
@@ -674,6 +680,11 @@ type StopEvent struct {
 	OccurredOn time.Time        `json:"occurred_on"`
 }
 
+type MsgDeletedEvent struct {
+	OrgID models.OrgID `json:"org_id"`
+	MsgID models.MsgID `json:"message_id"`
+}
+
 // creates a new event task for the passed in timeout event
 func newTimedTask(eventType string, orgID models.OrgID, contactID models.ContactID, sessionID models.SessionID, eventTime time.Time) *queue.Task {
 	event := &TimedEvent{

core/tasks/handler/handle_contact_event.go

@@ -131,6 +131,14 @@ func (t *HandleContactEventTask) Perform(ctx context.Context, rt *runtime.Runtim
 			}
 			err = handleTimedEvent(ctx, rt, contactEvent.Type, evt)
 
+		case MsgDeletedType:
+			evt := &MsgDeletedEvent{}
+			err = json.Unmarshal(contactEvent.Task, evt)
+			if err != nil {
+				return errors.Wrapf(err, "error unmarshalling deleted event: %s", event)
+			}
+			err = handleMsgDeletedEvent(ctx, rt, evt)
+
 		default:
 			return errors.Errorf("unknown contact event type: %s", contactEvent.Type)
 		}

testsuite/testsuite.go

@@ -266,6 +266,7 @@ DELETE FROM flows_flowrevision WHERE flow_id >= 30000;
 DELETE FROM flows_flow WHERE id >= 30000;
 DELETE FROM ivr_call;
 DELETE FROM campaigns_eventfire;
+DELETE FROM msgs_msg_labels;
 DELETE FROM msgs_msg;
 DELETE FROM msgs_broadcast_groups;
 DELETE FROM msgs_broadcast_contacts;