Merge pull request #668 from nyaruka/remove_topups

Remove topups

Author: rowanseymour

core/hooks/commit_ivr.go

@@ -24,22 +24,8 @@ func (h *commitIVRHook) Apply(ctx context.Context, rt *runtime.Runtime, tx *sqlx
 		}
 	}
 
-	// find the topup we will assign
-	topup, err := models.AllocateTopups(ctx, tx, rt.RP, oa.Org(), len(msgs))
-	if err != nil {
-		return errors.Wrapf(err, "error allocating topup for outgoing IVR message")
-	}
-
-	// if we have an active topup, assign it to our messages
-	if topup != models.NilTopupID {
-		for _, m := range msgs {
-			m.SetTopup(topup)
-		}
-	}
-
 	// insert all our messages
-	err = models.InsertMessages(ctx, tx, msgs)
-	if err != nil {
+	if err := models.InsertMessages(ctx, tx, msgs); err != nil {
 		return errors.Wrapf(err, "error writing messages")
 	}
 

core/hooks/commit_messages.go

@@ -24,22 +24,8 @@ func (h *commitMessagesHook) Apply(ctx context.Context, rt *runtime.Runtime, tx
 		}
 	}
 
-	// allocate a topup for this message if org uses topups
-	topup, err := models.AllocateTopups(ctx, tx, rt.RP, oa.Org(), len(msgs))
-	if err != nil {
-		return errors.Wrapf(err, "error allocating topup for outgoing message")
-	}
-
-	// if we have an active topup, assign it to our messages
-	if topup != models.NilTopupID {
-		for _, m := range msgs {
-			m.SetTopup(topup)
-		}
-	}
-
 	// insert all our messages
-	err = models.InsertMessages(ctx, tx, msgs)
-	if err != nil {
+	if err := models.InsertMessages(ctx, tx, msgs); err != nil {
 		return errors.Wrapf(err, "error writing messages")
 	}
 

core/ivr/ivr.go

@@ -580,17 +580,8 @@ func buildMsgResume(
 	// create an incoming message
 	msg := models.NewIncomingIVR(rt.Config, oa.OrgID(), call, msgIn, time.Now())
 
-	// allocate a topup for this message if org uses topups)
-	topupID, err := models.AllocateTopups(ctx, rt.DB, rt.RP, oa.Org(), 1)
-	if err != nil {
-		return nil, nil, errors.Wrapf(err, "error allocating topup for incoming IVR message")
-	}
-
-	msg.SetTopup(topupID)
-
 	// commit it
-	err = models.InsertMessages(ctx, rt.DB, []*models.Msg{msg})
-	if err != nil {
+	if err := models.InsertMessages(ctx, rt.DB, []*models.Msg{msg}); err != nil {
 		return nil, nil, errors.Wrapf(err, "error committing new message")
 	}
 

core/models/msgs.go

@@ -141,7 +141,6 @@ type Msg struct {
 		URN                  urns.URN           `db:"urn_urn"         json:"urn"`
 		URNAuth              null.String        `db:"urn_auth"        json:"urn_auth,omitempty"`
 		OrgID                OrgID              `db:"org_id"          json:"org_id"`
-		TopupID              TopupID            `db:"topup_id"        json:"-"`
 		FlowID               FlowID             `db:"flow_id"         json:"-"`
 
 		// extra data from handling added to the courier payload
@@ -184,14 +183,11 @@ func (m *Msg) ChannelUUID() assets.ChannelUUID  { return m.m.ChannelUUID }
 func (m *Msg) URN() urns.URN                    { return m.m.URN }
 func (m *Msg) URNAuth() null.String             { return m.m.URNAuth }
 func (m *Msg) OrgID() OrgID                     { return m.m.OrgID }
-func (m *Msg) TopupID() TopupID                 { return m.m.TopupID }
 func (m *Msg) FlowID() FlowID                   { return m.m.FlowID }
 func (m *Msg) ContactID() ContactID             { return m.m.ContactID }
 func (m *Msg) ContactURNID() *URNID             { return m.m.ContactURNID }
 func (m *Msg) IsResend() bool                   { return m.m.IsResend }
 
-func (m *Msg) SetTopup(topupID TopupID) { m.m.TopupID = topupID }
-
 func (m *Msg) SetChannel(channel *Channel) {
 	m.channel = channel
 	if channel != nil {
@@ -255,7 +251,6 @@ func NewIncomingIVR(cfg *runtime.Config, orgID OrgID, call *Call, in *flows.MsgI
 	m.ChannelID = call.ChannelID()
 
 	m.OrgID = orgID
-	m.TopupID = NilTopupID
 	m.CreatedOn = createdOn
 
 	// add any attachments
@@ -288,7 +283,6 @@ func NewOutgoingIVR(cfg *runtime.Config, orgID OrgID, call *Call, out *flows.Msg
 	m.URN = out.URN()
 
 	m.OrgID = orgID
-	m.TopupID = NilTopupID
 	m.CreatedOn = createdOn
 	m.SentOn = &createdOn
 
@@ -354,7 +348,6 @@ func newOutgoingMsg(rt *runtime.Runtime, org *Org, channel *Channel, contact *fl
 	m.OrgID = org.ID()
 	m.ContactID = ContactID(contact.ID())
 	m.BroadcastID = broadcastID
-	m.TopupID = NilTopupID
 	m.Text = out.Text()
 	m.HighPriority = false
 	m.Direction = DirectionOut
@@ -451,7 +444,6 @@ func NewIncomingMsg(cfg *runtime.Config, orgID OrgID, channel *Channel, contactI
 	m.MsgType = MsgTypeFlow
 	m.ContactID = contactID
 	m.OrgID = orgID
-	m.TopupID = NilTopupID
 	m.CreatedOn = createdOn
 
 	// add any attachments
@@ -483,8 +475,7 @@ SELECT
 	channel_id,
 	contact_id,
 	contact_urn_id,
-	org_id,
-	topup_id
+	org_id
 FROM
 	msgs_msg
 WHERE
@@ -521,7 +512,6 @@ SELECT
 	m.contact_id,
 	m.contact_urn_id,
 	m.org_id,
-	m.topup_id,
 	u.identity AS "urn_urn",
 	u.auth AS "urn_auth"
 FROM
@@ -626,18 +616,18 @@ const insertMsgSQL = `
 INSERT INTO
 msgs_msg(uuid, text, high_priority, created_on, modified_on, queued_on, sent_on, direction, status, attachments, metadata,
 		 visibility, msg_type, msg_count, error_count, next_attempt, failed_reason, channel_id,
-		 contact_id, contact_urn_id, org_id, topup_id, flow_id, broadcast_id)
+		 contact_id, contact_urn_id, org_id, flow_id, broadcast_id)
   VALUES(:uuid, :text, :high_priority, :created_on, now(), now(), :sent_on, :direction, :status, :attachments, :metadata,
 		 :visibility, :msg_type, :msg_count, :error_count, :next_attempt, :failed_reason, :channel_id,
-		 :contact_id, :contact_urn_id, :org_id, :topup_id, :flow_id, :broadcast_id)
+		 :contact_id, :contact_urn_id, :org_id, :flow_id, :broadcast_id)
 RETURNING 
 	id as id, 
 	now() as modified_on,
 	now() as queued_on
 `
 
 // UpdateMessage updates a message after handling
-func UpdateMessage(ctx context.Context, tx Queryer, msgID flows.MsgID, status MsgStatus, visibility MsgVisibility, msgType MsgType, flow FlowID, topup TopupID, attachments []utils.Attachment, logUUIDs []ChannelLogUUID) error {
+func UpdateMessage(ctx context.Context, tx Queryer, msgID flows.MsgID, status MsgStatus, visibility MsgVisibility, msgType MsgType, flow FlowID, attachments []utils.Attachment, logUUIDs []ChannelLogUUID) error {
 	_, err := tx.ExecContext(ctx,
 		`UPDATE 
 			msgs_msg 
@@ -646,12 +636,11 @@ func UpdateMessage(ctx context.Context, tx Queryer, msgID flows.MsgID, status Ms
 			visibility = $3,
 			msg_type = $4,
 			flow_id = $5,
-			topup_id = $6,
-			attachments = $7,
-			log_uuids = array_cat(log_uuids, $8)
+			attachments = $6,
+			log_uuids = array_cat(log_uuids, $7)
 		WHERE
 			id = $1`,
-		msgID, status, visibility, msgType, flow, topup, pq.Array(attachments), pq.Array(logUUIDs))
+		msgID, status, visibility, msgType, flow, pq.Array(attachments), pq.Array(logUUIDs))
 
 	if err != nil {
 		return errors.Wrapf(err, "error updating msg: %d", msgID)
@@ -1121,19 +1110,6 @@ func (b *BroadcastBatch) CreateMessages(ctx context.Context, rt *runtime.Runtime
 		}
 	}
 
-	// allocate a topup for these message if org uses topups
-	topup, err := AllocateTopups(ctx, rt.DB, rt.RP, oa.Org(), len(msgs))
-	if err != nil {
-		return nil, errors.Wrapf(err, "error allocating topup for broadcast messages")
-	}
-
-	// if we have an active topup, assign it to our messages
-	if topup != NilTopupID {
-		for _, m := range msgs {
-			m.SetTopup(topup)
-		}
-	}
-
 	// insert them in a single request
 	err = InsertMessages(ctx, rt.DB, msgs)
 	if err != nil {
@@ -1184,14 +1160,13 @@ func (b *BroadcastBatch) updateTicket(ctx context.Context, db Queryer, oa *OrgAs
 const sqlUpdateMsgForResending = `
 UPDATE msgs_msg m
    SET channel_id = r.channel_id::int,
-       topup_id = r.topup_id::int,
        status = 'P',
        error_count = 0,
        failed_reason = NULL,
        queued_on = r.queued_on::timestamp with time zone,
        sent_on = NULL,
        modified_on = NOW()
-  FROM (VALUES(:id, :channel_id, :topup_id, :queued_on)) AS r(id, channel_id, topup_id, queued_on)
+  FROM (VALUES(:id, :channel_id, :queued_on)) AS r(id, channel_id, queued_on)
  WHERE m.id = r.id::bigint`
 
 const sqlUpdateMsgResendFailed = `
@@ -1211,7 +1186,6 @@ func ResendMessages(ctx context.Context, db Queryer, rp *redis.Pool, oa *OrgAsse
 
 	for _, msg := range msgs {
 		var ch *flows.Channel
-		var err error
 		urnID := msg.ContactURNID()
 
 		if urnID != nil {
@@ -1243,11 +1217,6 @@ func ResendMessages(ctx context.Context, db Queryer, rp *redis.Pool, oa *OrgAsse
 			msg.m.FailedReason = ""
 			msg.m.IsResend = true // mark message as being a resend so it will be queued to courier as such
 
-			// allocate a new topup for this message if org uses topups
-			msg.m.TopupID, err = AllocateTopups(ctx, db, rp, oa.Org(), 1)
-			if err != nil {
-				return nil, errors.Wrapf(err, "error allocating topup for message resending")
-			}
 			resends = append(resends, msg.m)
 			resent = append(resent, msg)
 		} else {

core/models/msgs_test.go

@@ -472,13 +472,11 @@ func TestResendMessages(t *testing.T) {
 
 	assert.Len(t, resent, 3) // only #1, #2 and #3 can be resent
 
-	// both messages should now have a channel, a topup and be marked for resending
+	// both messages should now have a channel and be marked for resending
 	assert.True(t, resent[0].IsResend())
 	assert.Equal(t, testdata.TwilioChannel.ID, resent[0].ChannelID())
-	assert.Equal(t, models.TopupID(1), resent[0].TopupID())
 	assert.True(t, resent[1].IsResend())
 	assert.Equal(t, testdata.VonageChannel.ID, resent[1].ChannelID()) // channel changed
-	assert.Equal(t, models.TopupID(1), resent[1].TopupID())
 	assert.True(t, resent[2].IsResend())
 	assert.Equal(t, testdata.TwilioChannel.ID, resent[2].ChannelID()) // channel added
 

core/models/orgs.go

@@ -68,10 +68,9 @@ const (
 // Org is mailroom's type for RapidPro orgs. It also implements the envs.Environment interface for GoFlow
 type Org struct {
 	o struct {
-		ID         OrgID    `json:"id"`
-		Suspended  bool     `json:"is_suspended"`
-		UsesTopups bool     `json:"uses_topups"`
-		Config     null.Map `json:"config"`
+		ID        OrgID    `json:"id"`
+		Suspended bool     `json:"is_suspended"`
+		Config    null.Map `json:"config"`
 	}
 	env envs.Environment
 }
@@ -82,9 +81,6 @@ func (o *Org) ID() OrgID { return o.o.ID }
 // Suspended returns whether the org has been suspended
 func (o *Org) Suspended() bool { return o.o.Suspended }
 
-// UsesTopups returns whether the org uses topups
-func (o *Org) UsesTopups() bool { return o.o.UsesTopups }
-
 // DateFormat returns the date format for this org
 func (o *Org) DateFormat() envs.DateFormat { return o.env.DateFormat() }
 
@@ -252,7 +248,6 @@ const selectOrgByID = `
 SELECT ROW_TO_JSON(o) FROM (SELECT
 	id,
 	is_suspended,
-	uses_topups,
 	COALESCE(o.config::json,'{}'::json) AS config,
 	(SELECT CASE date_format WHEN 'D' THEN 'DD-MM-YYYY' WHEN 'M' THEN 'MM-DD-YYYY' END) AS date_format, 
 	'tt:mm' AS time_format,

core/models/orgs_test.go

@@ -36,7 +36,6 @@ func TestOrgs(t *testing.T) {
 
 	assert.Equal(t, models.OrgID(1), org.ID())
 	assert.False(t, org.Suspended())
-	assert.True(t, org.UsesTopups())
 	assert.Equal(t, envs.DateFormatDayMonthYear, org.DateFormat())
 	assert.Equal(t, envs.TimeFormatHourMinute, org.TimeFormat())
 	assert.Equal(t, envs.RedactionPolicyNone, org.RedactionPolicy())