diff --git a/CHANGELOG.md b/CHANGELOG.md index d6f00d3..692ca94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +v7.1.0 +---------- + * Remove msgs_msg.response_to_id + v7.0.0 ---------- * Test on PG12 and 13 diff --git a/archives/messages.go b/archives/messages.go index a3a5119..b957015 100644 --- a/archives/messages.go +++ b/archives/messages.go @@ -123,12 +123,6 @@ DELETE FROM msgs_msg_labels WHERE msg_id IN(?) ` -const unlinkResponses = ` -UPDATE msgs_msg -SET response_to_id = NULL -WHERE response_to_id IN(?) -` - const deleteMessages = ` DELETE FROM msgs_msg WHERE id IN(?) @@ -189,9 +183,7 @@ func DeleteArchivedMessages(ctx context.Context, config *Config, db *sqlx.DB, s3 } rows.Close() - log.WithFields(logrus.Fields{ - "msg_count": len(msgIDs), - }).Debug("found messages") + log.WithField("msg_count", len(msgIDs)).Debug("found messages") // verify we don't see more messages than there are in our archive (fewer is ok) if visibleCount > archive.RecordCount { @@ -199,19 +191,13 @@ func DeleteArchivedMessages(ctx context.Context, config *Config, db *sqlx.DB, s3 } // ok, delete our messages in batches, we do this in transactions as it spans a few different queries - for startIdx := 0; startIdx < len(msgIDs); startIdx += deleteTransactionSize { + for _, idBatch := range chunkIDs(msgIDs, deleteTransactionSize) { // no single batch should take more than a few minutes ctx, cancel := context.WithTimeout(ctx, time.Minute*15) defer cancel() start := time.Now() - endIdx := startIdx + deleteTransactionSize - if endIdx > len(msgIDs) { - endIdx = len(msgIDs) - } - batchIDs := msgIDs[startIdx:endIdx] - // start our transaction tx, err := db.BeginTxx(ctx, nil) if err != nil { @@ -219,45 +205,36 @@ func DeleteArchivedMessages(ctx context.Context, config *Config, db *sqlx.DB, s3 } // first update our delete_reason - err = executeInQuery(ctx, tx, setMessageDeleteReason, batchIDs) + err = executeInQuery(ctx, tx, setMessageDeleteReason, idBatch) if err != nil { - return fmt.Errorf("error updating delete reason: %s", err.Error()) + return errors.Wrap(err, "error updating delete reason") } // now delete any channel logs - err = executeInQuery(ctx, tx, deleteMessageLogs, batchIDs) + err = executeInQuery(ctx, tx, deleteMessageLogs, idBatch) if err != nil { - return fmt.Errorf("error removing channel logs: %s", err.Error()) + return errors.Wrap(err, "error removing channel logs") } // then any labels - err = executeInQuery(ctx, tx, deleteMessageLabels, batchIDs) + err = executeInQuery(ctx, tx, deleteMessageLabels, idBatch) if err != nil { - return fmt.Errorf("error removing message labels: %s", err.Error()) - } - - // unlink any responses - err = executeInQuery(ctx, tx, unlinkResponses, batchIDs) - if err != nil { - return fmt.Errorf("error unlinking responses: %s", err.Error()) + return errors.Wrap(err, "error removing message labels") } // finally, delete our messages - err = executeInQuery(ctx, tx, deleteMessages, batchIDs) + err = executeInQuery(ctx, tx, deleteMessages, idBatch) if err != nil { - return fmt.Errorf("error deleting messages: %s", err.Error()) + return errors.Wrap(err, "error deleting messages") } // commit our transaction err = tx.Commit() if err != nil { - return fmt.Errorf("error committing message delete transaction: %s", err.Error()) + return errors.Wrap(err, "error committing message delete transaction") } - log.WithFields(logrus.Fields{ - "elapsed": time.Since(start), - "count": len(batchIDs), - }).Debug("deleted batch of messages") + log.WithField("elapsed", time.Since(start)).WithField("count", len(idBatch)).Debug("deleted batch of messages") cancel() } @@ -270,14 +247,12 @@ func DeleteArchivedMessages(ctx context.Context, config *Config, db *sqlx.DB, s3 // all went well! mark our archive as no longer needing deletion _, err = db.ExecContext(outer, setArchiveDeleted, archive.ID, deletedOn) if err != nil { - return fmt.Errorf("error setting archive as deleted: %s", err.Error()) + return errors.Wrap(err, "error setting archive as deleted") } archive.NeedsDeletion = false archive.DeletedOn = &deletedOn - logrus.WithFields(logrus.Fields{ - "elapsed": time.Since(start), - }).Info("completed deleting messages") + logrus.WithField("elapsed", time.Since(start)).Info("completed deleting messages") return nil } diff --git a/archives/runs.go b/archives/runs.go index c2e314e..763e119 100644 --- a/archives/runs.go +++ b/archives/runs.go @@ -180,9 +180,7 @@ func DeleteArchivedRuns(ctx context.Context, config *Config, db *sqlx.DB, s3Clie } rows.Close() - log.WithFields(logrus.Fields{ - "run_count": len(runIDs), - }).Debug("found runs") + log.WithField("run_count", len(runIDs)).Debug("found runs") // verify we don't see more runs than there are in our archive (fewer is ok) if runCount > archive.RecordCount { @@ -190,19 +188,13 @@ func DeleteArchivedRuns(ctx context.Context, config *Config, db *sqlx.DB, s3Clie } // ok, delete our runs in batches, we do this in transactions as it spans a few different queries - for startIdx := 0; startIdx < len(runIDs); startIdx += deleteTransactionSize { + for _, idBatch := range chunkIDs(runIDs, deleteTransactionSize) { // no single batch should take more than a few minutes ctx, cancel := context.WithTimeout(ctx, time.Minute*15) defer cancel() start := time.Now() - endIdx := startIdx + deleteTransactionSize - if endIdx > len(runIDs) { - endIdx = len(runIDs) - } - batchIDs := runIDs[startIdx:endIdx] - // start our transaction tx, err := db.BeginTxx(ctx, nil) if err != nil { @@ -210,39 +202,36 @@ func DeleteArchivedRuns(ctx context.Context, config *Config, db *sqlx.DB, s3Clie } // first update our delete_reason - err = executeInQuery(ctx, tx, setRunDeleteReason, batchIDs) + err = executeInQuery(ctx, tx, setRunDeleteReason, idBatch) if err != nil { - return fmt.Errorf("error updating delete reason: %s", err.Error()) + return errors.Wrap(err, "error updating delete reason") } // any recent runs - err = executeInQuery(ctx, tx, deleteRecentRuns, batchIDs) + err = executeInQuery(ctx, tx, deleteRecentRuns, idBatch) if err != nil { - return fmt.Errorf("error deleting recent runs: %s", err.Error()) + return errors.Wrap(err, "error deleting recent runs") } // unlink any parents - err = executeInQuery(ctx, tx, unlinkParents, batchIDs) + err = executeInQuery(ctx, tx, unlinkParents, idBatch) if err != nil { - return fmt.Errorf("error unliking parent runs: %s", err.Error()) + return errors.Wrap(err, "error unliking parent runs") } // finally, delete our runs - err = executeInQuery(ctx, tx, deleteRuns, batchIDs) + err = executeInQuery(ctx, tx, deleteRuns, idBatch) if err != nil { - return fmt.Errorf("error deleting runs: %s", err.Error()) + return errors.Wrap(err, "error deleting runs") } // commit our transaction err = tx.Commit() if err != nil { - return fmt.Errorf("error committing run delete transaction: %s", err.Error()) + return errors.Wrap(err, "error committing run delete transaction") } - log.WithFields(logrus.Fields{ - "elapsed": time.Since(start), - "count": len(batchIDs), - }).Debug("deleted batch of runs") + log.WithField("elapsed", time.Since(start)).WithField("count", len(idBatch)).Debug("deleted batch of runs") cancel() } @@ -255,14 +244,12 @@ func DeleteArchivedRuns(ctx context.Context, config *Config, db *sqlx.DB, s3Clie // all went well! mark our archive as no longer needing deletion _, err = db.ExecContext(outer, setArchiveDeleted, archive.ID, deletedOn) if err != nil { - return fmt.Errorf("error setting archive as deleted: %s", err.Error()) + return errors.Wrap(err, "error setting archive as deleted") } archive.NeedsDeletion = false archive.DeletedOn = &deletedOn - logrus.WithFields(logrus.Fields{ - "elapsed": time.Since(start), - }).Info("completed deleting runs") + logrus.WithField("elapsed", time.Since(start)).Info("completed deleting runs") return nil } diff --git a/archives/utils.go b/archives/utils.go new file mode 100644 index 0000000..d50db23 --- /dev/null +++ b/archives/utils.go @@ -0,0 +1,15 @@ +package archives + +// chunks a slice of in64 IDs +func chunkIDs(ids []int64, size int) [][]int64 { + chunks := make([][]int64, 0, len(ids)/size+1) + + for i := 0; i < len(ids); i += size { + end := i + size + if end > len(ids) { + end = len(ids) + } + chunks = append(chunks, ids[i:end]) + } + return chunks +} diff --git a/testdb.sql b/testdb.sql index c4bdb2e..dab01e5 100644 --- a/testdb.sql +++ b/testdb.sql @@ -90,8 +90,7 @@ CREATE TABLE msgs_msg ( org_id integer NOT NULL references orgs_org(id) on delete cascade, metadata text, topup_id integer, - delete_reason char(1) NULL, - response_to_id integer NULL references msgs_msg(id) + delete_reason char(1) NULL ); DROP TABLE IF EXISTS msgs_broadcast_recipients; @@ -269,15 +268,15 @@ INSERT INTO msgs_broadcast(id, text, created_on, purged, org_id, schedule_id) VA (3, 'base=>"not purged"'::hstore, '2017-08-12 19:11:59.890662+02:00', FALSE, 2, NULL), (4, 'base=>"new"'::hstore, '2019-08-12 19:11:59.890662+02:00', FALSE, 2, NULL); -INSERT INTO msgs_msg(id, broadcast_id, uuid, text, created_on, sent_on, modified_on, direction, status, visibility, msg_type, attachments, channel_id, contact_id, contact_urn_id, org_id, msg_count, error_count, next_attempt, response_to_id) VALUES -(1, NULL, '2f969340-704a-4aa2-a1bd-2f832a21d257', 'message 1', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-08-12 21:11:59.890662+00', NULL), -(2, NULL, 'abe87ac1-015c-4803-be29-1e89509fe682', 'message 2', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'I', 'H', 'D', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-08-12 21:11:59.890662+00', NULL), -(3, NULL, 'a7e83a22-a6ff-4e18-82d0-19545640ccba', 'message 3', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'O', 'H', 'V', 'I', '{"image/png:https://foo.bar/image1.png", "image/png:https://foo.bar/image2.png"}', NULL, 6, 7, 2, 1, 0, '2017-08-12 21:11:59.890662+00', NULL), -(4, NULL, '1cad36af-5581-4c8a-81cd-83708398f61e', 'message 4', '2017-08-13 21:11:59.890662+00', '2017-08-13 21:11:59.890662+00', '2017-08-13 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-08-13 21:11:59.890662+00', NULL), -(5, NULL, 'f557972e-2eb5-42fa-9b87-902116d18787', 'message 5', '2017-08-11 21:11:59.890662+02:00', '2017-08-11 21:11:59.890662+02:00', '2017-08-11 21:11:59.890662+02:00', 'I', 'H', 'V', 'I', NULL, 3, 7, 8, 3, 1, 0, '2017-08-11 21:11:59.890662+02:00', NULL), -(6, 2, '579d148c-0ab1-4afb-832f-afb1fe0e19b7', 'message 6', '2017-10-08 21:11:59.890662+00', '2017-10-08 21:11:59.890662+00', '2017-10-08 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-10-08 21:11:59.890662+00', NULL), -(7, NULL, '7aeca469-2593-444e-afe4-4702317534c9', 'message 7', '2018-01-02 21:11:59.890662+00', '2018-01-02 21:11:59.890662+00', '2018-01-02 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2018-01-02 21:11:59.890662+00', 2), -(9, NULL, 'e14ab466-0d3b-436d-a0f7-5851fd7d9b7d', 'message 9', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'O', 'S', 'V', 'F', NULL, NULL, 6, NULL, 2, 1, 0, '2017-08-12 21:11:59.890662+00', NULL); +INSERT INTO msgs_msg(id, broadcast_id, uuid, text, created_on, sent_on, modified_on, direction, status, visibility, msg_type, attachments, channel_id, contact_id, contact_urn_id, org_id, msg_count, error_count, next_attempt) VALUES +(1, NULL, '2f969340-704a-4aa2-a1bd-2f832a21d257', 'message 1', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-08-12 21:11:59.890662+00'), +(2, NULL, 'abe87ac1-015c-4803-be29-1e89509fe682', 'message 2', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'I', 'H', 'D', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-08-12 21:11:59.890662+00'), +(3, NULL, 'a7e83a22-a6ff-4e18-82d0-19545640ccba', 'message 3', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'O', 'H', 'V', 'I', '{"image/png:https://foo.bar/image1.png", "image/png:https://foo.bar/image2.png"}', NULL, 6, 7, 2, 1, 0, '2017-08-12 21:11:59.890662+00'), +(4, NULL, '1cad36af-5581-4c8a-81cd-83708398f61e', 'message 4', '2017-08-13 21:11:59.890662+00', '2017-08-13 21:11:59.890662+00', '2017-08-13 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-08-13 21:11:59.890662+00'), +(5, NULL, 'f557972e-2eb5-42fa-9b87-902116d18787', 'message 5', '2017-08-11 21:11:59.890662+02:00', '2017-08-11 21:11:59.890662+02:00', '2017-08-11 21:11:59.890662+02:00', 'I', 'H', 'V', 'I', NULL, 3, 7, 8, 3, 1, 0, '2017-08-11 21:11:59.890662+02:00'), +(6, 2, '579d148c-0ab1-4afb-832f-afb1fe0e19b7', 'message 6', '2017-10-08 21:11:59.890662+00', '2017-10-08 21:11:59.890662+00', '2017-10-08 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2017-10-08 21:11:59.890662+00'), +(7, NULL, '7aeca469-2593-444e-afe4-4702317534c9', 'message 7', '2018-01-02 21:11:59.890662+00', '2018-01-02 21:11:59.890662+00', '2018-01-02 21:11:59.890662+00', 'I', 'H', 'V', 'I', NULL, 2, 6, 7, 2, 1, 0, '2018-01-02 21:11:59.890662+00'), +(9, NULL, 'e14ab466-0d3b-436d-a0f7-5851fd7d9b7d', 'message 9', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', '2017-08-12 21:11:59.890662+00', 'O', 'S', 'V', 'F', NULL, NULL, 6, NULL, 2, 1, 0, '2017-08-12 21:11:59.890662+00'); INSERT INTO msgs_label(id, uuid, name) VALUES (1, '1d9e3188-b74b-4ae0-a166-0de31aedb34a', 'Label 1'),