Merge branch 'main' into remove_topups

Author: rowanseymour

CHANGELOG.md

@@ -1,3 +1,11 @@
+v7.5.15
+----------
+ * Update to latest goflow which changes expirations for dial waits
+ * Add support for time limits on dial waits
+ * Rework message events to use channel UUID and include channel type
+ * Simplify getting active call count
+ * Allow incoming call triggers to take message flows
+
 v7.5.14
 ----------
  * Switch to new ivr_call table

core/ivr/ivr.go

@@ -68,9 +68,8 @@ type Service interface {
 	HangupCall(externalID string) (*httpx.Trace, error)
 
 	WriteSessionResponse(ctx context.Context, rt *runtime.Runtime, channel *models.Channel, call *models.Call, session *models.Session, number urns.URN, resumeURL string, req *http.Request, w http.ResponseWriter) error
-
+	WriteHangupResponse(w http.ResponseWriter) error
 	WriteErrorResponse(w http.ResponseWriter, err error) error
-
 	WriteEmptyResponse(w http.ResponseWriter, msg string) error
 
 	ResumeForRequest(r *http.Request) (Resume, error)

core/tasks/handler/cron.go

@@ -105,22 +105,23 @@ func RetryPendingMsgs(ctx context.Context, rt *runtime.Runtime) error {
 
 const unhandledMsgsQuery = `
 SELECT org_id, contact_id, msg_id, ROW_TO_JSON(r) FROM (SELECT
-	m.contact_id as contact_id,
-	m.org_id as org_id, 
-	m.channel_id as channel_id,
-	m.id as msg_id,
-	m.uuid as msg_uuid,
-	m.external_id as msg_external_id,
-	u.identity as urn,
-	m.contact_urn_id as urn_id,
-	m.text as text,
-	m.attachments as attachments
+	m.contact_id AS contact_id,
+	m.org_id AS org_id, 
+	c.id AS channel_id,
+	c.uuid AS channel_uuid,
+	c.channel_type AS channel_type,
+	m.id AS msg_id,
+	m.uuid AS msg_uuid,
+	m.external_id AS msg_external_id,
+	u.identity AS urn,
+	m.contact_urn_id AS urn_id,
+	m.text AS text,
+	m.attachments AS attachments
 FROM
 	msgs_msg m
-	JOIN contacts_contacturn as u ON m.contact_urn_id = u.id
+	INNER JOIN channels_channel c ON c.id = m.channel_id 
+	INNER JOIN contacts_contacturn u ON u.id = m.contact_urn_id
 WHERE
-	m.direction = 'I' AND
-	m.status = 'P' AND
-	m.created_on < now() - INTERVAL '5 min'
+	m.direction = 'I' AND m.status = 'P' AND m.created_on < now() - INTERVAL '5 min'
 ) r;
 `

core/tasks/handler/handler_test.go

@@ -300,14 +300,15 @@ func TestMsgEvents(t *testing.T) {
 
 	makeMsgTask := func(org *testdata.Org, channel *testdata.Channel, contact *testdata.Contact, text string) *queue.Task {
 		return &queue.Task{Type: handler.MsgEventType, OrgID: int(org.ID), Task: jsonx.MustMarshal(&handler.MsgEvent{
-			ContactID: contact.ID,
-			OrgID:     org.ID,
-			ChannelID: channel.ID,
-			MsgID:     dbMsg.ID(),
-			MsgUUID:   dbMsg.UUID(),
-			URN:       contact.URN,
-			URNID:     contact.URNID,
-			Text:      text,
+			ContactID:   contact.ID,
+			OrgID:       org.ID,
+			ChannelUUID: channel.UUID,
+			ChannelType: channel.Type,
+			MsgID:       dbMsg.ID(),
+			MsgUUID:     dbMsg.UUID(),
+			URN:         contact.URN,
+			URNID:       contact.URNID,
+			Text:        text,
 		})}
 	}
 
@@ -587,47 +588,47 @@ func TestTimedEvents(t *testing.T) {
 		Contact   *testdata.Contact
 		Message   string
 		Response  string
-		ChannelID models.ChannelID
-		OrgID     models.OrgID
+		Channel   *testdata.Channel
+		Org       *testdata.Org
 	}{
 		// 0: start the flow
-		{handler.MsgEventType, testdata.Cathy, "start", "What is your favorite color?", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "start", "What is your favorite color?", testdata.TwitterChannel, testdata.Org1},
 
 		// 1: this expiration does nothing because the times don't match
-		{handler.ExpirationEventType, testdata.Cathy, "bad", "", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.ExpirationEventType, testdata.Cathy, "bad", "", testdata.TwitterChannel, testdata.Org1},
 
 		// 2: this checks that the flow wasn't expired
-		{handler.MsgEventType, testdata.Cathy, "red", "Good choice, I like Red too! What is your favorite beer?", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "red", "Good choice, I like Red too! What is your favorite beer?", testdata.TwitterChannel, testdata.Org1},
 
 		// 3: this expiration will actually take
-		{handler.ExpirationEventType, testdata.Cathy, "good", "", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.ExpirationEventType, testdata.Cathy, "good", "", testdata.TwitterChannel, testdata.Org1},
 
 		// 4: we won't get a response as we will be out of the flow
-		{handler.MsgEventType, testdata.Cathy, "mutzig", "", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "mutzig", "", testdata.TwitterChannel, testdata.Org1},
 
 		// 5: start the parent expiration flow
-		{handler.MsgEventType, testdata.Cathy, "parent", "Child", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "parent", "Child", testdata.TwitterChannel, testdata.Org1},
 
 		// 6: respond, should bring us out
-		{handler.MsgEventType, testdata.Cathy, "hi", "Completed", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "hi", "Completed", testdata.TwitterChannel, testdata.Org1},
 
 		// 7: expiring our child should be a no op
-		{handler.ExpirationEventType, testdata.Cathy, "child", "", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.ExpirationEventType, testdata.Cathy, "child", "", testdata.TwitterChannel, testdata.Org1},
 
 		// 8: respond one last time, should be done
-		{handler.MsgEventType, testdata.Cathy, "done", "Ended", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "done", "Ended", testdata.TwitterChannel, testdata.Org1},
 
 		// 9: start our favorite flow again
-		{handler.MsgEventType, testdata.Cathy, "start", "What is your favorite color?", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "start", "What is your favorite color?", testdata.TwitterChannel, testdata.Org1},
 
 		// 10: timeout on the color question
-		{handler.TimeoutEventType, testdata.Cathy, "", "Sorry you can't participate right now, I'll try again later.", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.TimeoutEventType, testdata.Cathy, "", "Sorry you can't participate right now, I'll try again later.", testdata.TwitterChannel, testdata.Org1},
 
 		// 11: start the pick a number flow
-		{handler.MsgEventType, testdata.Cathy, "pick", "Pick a number between 1-10.", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.MsgEventType, testdata.Cathy, "pick", "Pick a number between 1-10.", testdata.TwitterChannel, testdata.Org1},
 
 		// 12: try to resume with timeout even tho flow doesn't have one set
-		{handler.TimeoutEventType, testdata.Cathy, "", "", testdata.TwitterChannel.ID, testdata.Org1.ID},
+		{handler.TimeoutEventType, testdata.Cathy, "", "", testdata.TwitterChannel, testdata.Org1},
 	}
 
 	last := time.Now()
@@ -642,16 +643,17 @@ func TestTimedEvents(t *testing.T) {
 		if tc.EventType == handler.MsgEventType {
 			task = &queue.Task{
 				Type:  tc.EventType,
-				OrgID: int(tc.OrgID),
+				OrgID: int(tc.Org.ID),
 				Task: jsonx.MustMarshal(&handler.MsgEvent{
-					ContactID: tc.Contact.ID,
-					OrgID:     tc.OrgID,
-					ChannelID: tc.ChannelID,
-					MsgID:     flows.MsgID(1),
-					MsgUUID:   flows.MsgUUID(uuids.New()),
-					URN:       tc.Contact.URN,
-					URNID:     tc.Contact.URNID,
-					Text:      tc.Message,
+					ContactID:   tc.Contact.ID,
+					OrgID:       tc.Org.ID,
+					ChannelUUID: tc.Channel.UUID,
+					ChannelType: tc.Channel.Type,
+					MsgID:       flows.MsgID(1),
+					MsgUUID:     flows.MsgUUID(uuids.New()),
+					URN:         tc.Contact.URN,
+					URNID:       tc.Contact.URNID,
+					Text:        tc.Message,
 				}),
 			}
 		} else if tc.EventType == handler.ExpirationEventType {
@@ -666,15 +668,15 @@ func TestTimedEvents(t *testing.T) {
 				expiration = time.Now().Add(time.Hour * 24)
 			}
 
-			task = handler.NewExpirationTask(tc.OrgID, tc.Contact.ID, sessionID, expiration)
+			task = handler.NewExpirationTask(tc.Org.ID, tc.Contact.ID, sessionID, expiration)
 
 		} else if tc.EventType == handler.TimeoutEventType {
 			timeoutOn := time.Now().Round(time.Millisecond) // so that there's no difference between this and what we read from the db
 
 			// usually courier will set timeout_on after sending the last message
 			db.MustExec(`UPDATE flows_flowsession SET timeout_on = $2 WHERE id = $1`, sessionID, timeoutOn)
 
-			task = handler.NewTimeoutTask(tc.OrgID, tc.Contact.ID, sessionID, timeoutOn)
+			task = handler.NewTimeoutTask(tc.Org.ID, tc.Contact.ID, sessionID, timeoutOn)
 		}
 
 		err := handler.QueueHandleTask(rc, tc.Contact.ID, task)

core/tasks/handler/worker.go

@@ -11,6 +11,7 @@ import (
 	"github.com/nyaruka/gocommon/analytics"
 	"github.com/nyaruka/gocommon/dbutil"
 	"github.com/nyaruka/gocommon/urns"
+	"github.com/nyaruka/goflow/assets"
 	"github.com/nyaruka/goflow/excellent/types"
 	"github.com/nyaruka/goflow/flows"
 	"github.com/nyaruka/goflow/flows/engine"
@@ -382,7 +383,7 @@ func HandleChannelEvent(ctx context.Context, rt *runtime.Runtime, eventType mode
 		return nil, errors.Wrapf(err, "error loading flow for trigger")
 	}
 
-	// if this is an IVR flow, we need to trigger that start (which happens in a different queue)
+	// if this is an IVR flow and we don't have a call, trigger that asynchronously
 	if flow.FlowType() == models.FlowTypeVoice && call == nil {
 		err = runner.TriggerIVRFlow(ctx, rt, oa.OrgID(), flow.ID(), []models.ContactID{modelContact.ID()}, nil)
 		if err != nil {
@@ -428,7 +429,7 @@ func HandleChannelEvent(ctx context.Context, rt *runtime.Runtime, eventType mode
 	// if we have a channel connection we set the connection on the session before our event hooks fire
 	// so that IVR messages can be created with the right connection reference
 	var hook models.SessionCommitHook
-	if call != nil {
+	if flow.FlowType() == models.FlowTypeVoice && call != nil {
 		hook = func(ctx context.Context, tx *sqlx.Tx, rp *redis.Pool, oa *models.OrgAssets, sessions []*models.Session) error {
 			for _, session := range sessions {
 				session.SetCall(call)
@@ -498,7 +499,7 @@ func handleMsgEvent(ctx context.Context, rt *runtime.Runtime, event *MsgEvent) e
 	modelContact := contacts[0]
 
 	// load the channel for this message
-	channel := oa.ChannelByID(event.ChannelID)
+	channel := oa.ChannelByUUID(event.ChannelUUID)
 
 	// if we have URNs make sure the message URN is our highest priority (this is usually a noop)
 	if len(modelContact.URNs()) > 0 {
@@ -792,7 +793,8 @@ type TimedEvent struct {
 type MsgEvent struct {
 	ContactID     models.ContactID   `json:"contact_id"`
 	OrgID         models.OrgID       `json:"org_id"`
-	ChannelID     models.ChannelID   `json:"channel_id"`
+	ChannelUUID   assets.ChannelUUID `json:"channel_uuid"`
+	ChannelType   models.ChannelType `json:"channel_type"`
 	MsgID         flows.MsgID        `json:"msg_id"`
 	MsgUUID       flows.MsgUUID      `json:"msg_uuid"`
 	MsgExternalID null.String        `json:"msg_external_id"`

core/tasks/ivr/worker_test.go

@@ -92,6 +92,10 @@ func (s *MockService) WriteSessionResponse(ctx context.Context, rt *runtime.Runt
 	return nil
 }
 
+func (s *MockService) WriteHangupResponse(w http.ResponseWriter) error {
+	return nil
+}
+
 func (s *MockService) WriteErrorResponse(w http.ResponseWriter, err error) error {
 	return nil
 }

go.mod

@@ -15,8 +15,8 @@ require (
 	github.com/jmoiron/sqlx v1.3.5
 	github.com/lib/pq v1.10.6
 	github.com/nyaruka/ezconf v0.2.1
-	github.com/nyaruka/gocommon v1.30.0
-	github.com/nyaruka/goflow v0.171.0
+	github.com/nyaruka/gocommon v1.30.2
+	github.com/nyaruka/goflow v0.172.2
 	github.com/nyaruka/logrus_sentry v0.8.2-0.20190129182604-c2962b80ba7d
 	github.com/nyaruka/null v1.2.0
 	github.com/nyaruka/redisx v0.2.2

go.sum

@@ -218,10 +218,10 @@ github.com/naoina/toml v0.1.1 h1:PT/lllxVVN0gzzSqSlHEmP8MJB4MY2U7STGxiouV4X8=
 github.com/naoina/toml v0.1.1/go.mod h1:NBIhNtsFMo3G2szEBne+bO4gS192HuIYRqfvOWb4i1E=
 github.com/nyaruka/ezconf v0.2.1 h1:TDXWoqjqYya1uhou1mAJZg7rgFYL98EB0Tb3+BWtUh0=
 github.com/nyaruka/ezconf v0.2.1/go.mod h1:ey182kYkw2MIi4XiWe1FR/mzI33WCmTWuceDYYxgnQw=
-github.com/nyaruka/gocommon v1.30.0 h1:394CU1fxGXSs+mrd5x8nXnqw2epT9P0ui/9MtSE2ycQ=
-github.com/nyaruka/gocommon v1.30.0/go.mod h1:PApT/06fP5Tzs4/kbkJ+rVoyOc9Lbqm1lR0ow8Vqzp0=
-github.com/nyaruka/goflow v0.171.0 h1:2wjKClK8Z5kjZUaTSDGg+OihFgHf5PkylbvZM3FFFLI=
-github.com/nyaruka/goflow v0.171.0/go.mod h1:BELqxxpx4dqpBvaqM2TSXyq3rdJvX4RPlSAyuevjL+4=
+github.com/nyaruka/gocommon v1.30.2 h1:tWpq0jOfwKLCkN8LdE8ZGs22l2OYpTQAvdjAW6scl/Y=
+github.com/nyaruka/gocommon v1.30.2/go.mod h1:PApT/06fP5Tzs4/kbkJ+rVoyOc9Lbqm1lR0ow8Vqzp0=
+github.com/nyaruka/goflow v0.172.2 h1:wGpVeuiNHzP7Qx8Q7p8HalDq/RrIIpdylrD15rBYbTg=
+github.com/nyaruka/goflow v0.172.2/go.mod h1:BELqxxpx4dqpBvaqM2TSXyq3rdJvX4RPlSAyuevjL+4=
 github.com/nyaruka/librato v1.0.0 h1:Vznj9WCeC1yZXbBYyYp40KnbmXLbEkjKmHesV/v2SR0=
 github.com/nyaruka/librato v1.0.0/go.mod h1:pkRNLFhFurOz0QqBz6/DuTFhHHxAubWxs4Jx+J7yUgg=
 github.com/nyaruka/logrus_sentry v0.8.2-0.20190129182604-c2962b80ba7d h1:hyp9u36KIwbTCo2JAJ+TuJcJBc+UZzEig7RI/S5Dvkc=

services/ivr/twiml/client.go

@@ -24,10 +24,11 @@ type Redirect struct {
 }
 
 type Dial struct {
-	XMLName string `xml:"Dial"`
-	Number  string `xml:",chardata"`
-	Action  string `xml:"action,attr"`
-	Timeout int    `xml:"timeout,attr,omitempty"`
+	XMLName   string `xml:"Dial"`
+	Number    string `xml:",chardata"`
+	Action    string `xml:"action,attr"`
+	Timeout   int    `xml:"timeout,attr,omitempty"`
+	TimeLimit int    `xml:"timeLimit,attr,omitempty"`
 }
 
 type Gather struct {

services/ivr/twiml/service.go

@@ -376,29 +376,37 @@ func (s *service) WriteSessionResponse(ctx context.Context, rt *runtime.Runtime,
 	return nil
 }
 
+func (s *service) WriteHangupResponse(w http.ResponseWriter) error {
+	return s.writeResponse(w, &Response{
+		Commands: []any{Hangup{}},
+	})
+}
+
 // WriteErrorResponse writes an error / unavailable response
 func (s *service) WriteErrorResponse(w http.ResponseWriter, err error) error {
-	r := &Response{Message: strings.Replace(err.Error(), "--", "__", -1)}
-	r.Commands = append(r.Commands, Say{Text: ivr.ErrorMessage})
-	r.Commands = append(r.Commands, Hangup{})
-
-	body, err := xml.Marshal(r)
-	if err != nil {
-		return err
-	}
-	_, err = w.Write([]byte(xml.Header + string(body)))
-	return err
+	return s.writeResponse(w, &Response{
+		Message: strings.Replace(err.Error(), "--", "__", -1),
+		Commands: []any{
+			Say{Text: ivr.ErrorMessage},
+			Hangup{},
+		},
+	})
 }
 
 // WriteEmptyResponse writes an empty (but valid) response
 func (s *service) WriteEmptyResponse(w http.ResponseWriter, msg string) error {
-	r := &Response{Message: strings.Replace(msg, "--", "__", -1)}
+	return s.writeResponse(w, &Response{
+		Message: strings.Replace(msg, "--", "__", -1),
+	})
+}
 
-	body, err := xml.Marshal(r)
+func (s *service) writeResponse(w http.ResponseWriter, resp *Response) error {
+	marshalled, err := xml.Marshal(resp)
 	if err != nil {
 		return err
 	}
-	_, err = w.Write([]byte(xml.Header + string(body)))
+	w.Write([]byte(xml.Header))
+	_, err = w.Write(marshalled)
 	return err
 }
 
@@ -496,7 +504,7 @@ func ResponseForSprint(cfg *runtime.Config, number urns.URN, resumeURL string, e
 
 		case *events.DialWaitEvent:
 			hasWait = true
-			dial := Dial{Action: resumeURL + "&wait_type=dial", Number: event.URN.Path()}
+			dial := Dial{Action: resumeURL + "&wait_type=dial", Number: event.URN.Path(), Timeout: event.DialLimitSeconds, TimeLimit: event.CallLimitSeconds}
 			commands = append(commands, dial)
 			r.Commands = commands
 		}

services/ivr/twiml/service_test.go

@@ -98,9 +98,9 @@ func TestResponseForSprint(t *testing.T) {
 		},
 		{
 			[]flows.Event{
-				events.NewDialWait(urns.URN(`tel:+1234567890`), &expiresOn),
+				events.NewDialWait(urns.URN(`tel:+1234567890`), 60, 7200, &expiresOn),
 			},
-			`<Response><Dial action="http://temba.io/resume?session=1&amp;wait_type=dial">+1234567890</Dial></Response>`,
+			`<Response><Dial action="http://temba.io/resume?session=1&amp;wait_type=dial" timeout="60" timeLimit="7200">+1234567890</Dial></Response>`,
 		},
 	}
 

services/ivr/vonage/client.go

@@ -24,16 +24,18 @@ type CallRequest struct {
 
 	NCCO             []NCCO `json:"ncco,omitempty"`
 	MachineDetection string `json:"machine_detection"`
+	LengthTimer      int    `json:"length_timer,omitempty"`
 	RingingTimer     int    `json:"ringing_timer,omitempty"`
 }
 
 // CallResponse is the response from creating a new call
-// {
-//  "uuid": "63f61863-4a51-4f6b-86e1-46edebcf9356",
-//  "status": "started",
-//  "direction": "outbound",
-//  "conversation_uuid": "CON-f972836a-550f-45fa-956c-12a2ab5b7d22"
-// }
+//
+//	{
+//	 "uuid": "63f61863-4a51-4f6b-86e1-46edebcf9356",
+//	 "status": "started",
+//	 "direction": "outbound",
+//	 "conversation_uuid": "CON-f972836a-550f-45fa-956c-12a2ab5b7d22"
+//	}
 type CallResponse struct {
 	UUID             string `json:"uuid"`
 	Status           string `json:"status"`