fix(filters): max downloads per filter check (#660)

internal/action/run.go

@@ -87,6 +87,7 @@ func (s *service) RunAction(ctx context.Context, action *domain.Action, release
 		Type:       action.Type,
 		Client:     action.Client.Name,
 		Filter:     release.Filter.Name,
+		FilterID:   int64(release.Filter.ID),
 		Rejections: []string{},
 		Timestamp:  time.Now(),
 	}

internal/database/filter.go

@@ -1042,13 +1042,13 @@ func (r *FilterRepo) attachDownloadsByFilter(ctx context.Context, tx *Tx, filter
 
 func (r *FilterRepo) downloadsByFilterSqlite(ctx context.Context, tx *Tx, filterID int) (*domain.FilterDownloads, error) {
 	query := `SELECT
-    IFNULL(SUM(CASE WHEN "release".timestamp >= strftime('%Y-%m-%dT%H:00:00Z', datetime('now','localtime')) THEN 1 ELSE 0 END),0) as "hour_count",
+    IFNULL(SUM(CASE WHEN release_action_status.timestamp >= strftime('%Y-%m-%d %H:00:00', datetime('now','localtime')) THEN 1 ELSE 0 END),0) as "hour_count",
-    IFNULL(SUM(CASE WHEN "release".timestamp >= datetime('now', 'localtime', 'start of day') THEN 1 ELSE 0 END),0) as "day_count",
+    IFNULL(SUM(CASE WHEN release_action_status.timestamp >= datetime('now', 'localtime', 'start of day') THEN 1 ELSE 0 END),0) as "day_count",
-    IFNULL(SUM(CASE WHEN "release".timestamp >= datetime('now', 'localtime', 'weekday 0', '-7 days') THEN 1 ELSE 0 END),0) as "week_count",
+    IFNULL(SUM(CASE WHEN release_action_status.timestamp >= datetime('now', 'localtime', 'weekday 0', '-7 days') THEN 1 ELSE 0 END),0) as "week_count",
-    IFNULL(SUM(CASE WHEN "release".timestamp >= datetime('now', 'localtime', 'start of month') THEN 1 ELSE 0 END),0) as "month_count",
+    IFNULL(SUM(CASE WHEN release_action_status.timestamp >= datetime('now', 'localtime', 'start of month') THEN 1 ELSE 0 END),0) as "month_count",
     count(*) as "total_count"
-FROM "release"
+FROM release_action_status
-WHERE "release".filter_id = ?;`
+WHERE release_action_status.status = 'PUSH_APPROVED' AND release_action_status.filter_id = ?;`
 
 	row := tx.QueryRowContext(ctx, query, filterID)
 	if err := row.Err(); err != nil {
@@ -1068,13 +1068,13 @@ WHERE "release".filter_id = ?;`
 
 func (r *FilterRepo) downloadsByFilterPostgres(ctx context.Context, tx *Tx, filterID int) (*domain.FilterDownloads, error) {
 	query := `SELECT
-    COALESCE(SUM(CASE WHEN "release".timestamp >= date_trunc('hour', CURRENT_TIMESTAMP) THEN 1 ELSE 0 END),0) as "hour_count",
+    COALESCE(SUM(CASE WHEN release_action_status.timestamp >= date_trunc('hour', CURRENT_TIMESTAMP) THEN 1 ELSE 0 END),0) as "hour_count",
-    COALESCE(SUM(CASE WHEN "release".timestamp >= date_trunc('day', CURRENT_DATE) THEN 1 ELSE 0 END),0) as "day_count",
+    COALESCE(SUM(CASE WHEN release_action_status.timestamp >= date_trunc('day', CURRENT_DATE) THEN 1 ELSE 0 END),0) as "day_count",
-    COALESCE(SUM(CASE WHEN "release".timestamp >= date_trunc('week', CURRENT_DATE) THEN 1 ELSE 0 END),0) as "week_count",
+    COALESCE(SUM(CASE WHEN release_action_status.timestamp >= date_trunc('week', CURRENT_DATE) THEN 1 ELSE 0 END),0) as "week_count",
-    COALESCE(SUM(CASE WHEN "release".timestamp >= date_trunc('month', CURRENT_DATE) THEN 1 ELSE 0 END),0) as "month_count",
+    COALESCE(SUM(CASE WHEN release_action_status.timestamp >= date_trunc('month', CURRENT_DATE) THEN 1 ELSE 0 END),0) as "month_count",
     count(*) as "total_count"
-FROM "release"
+FROM release_action_status
-WHERE "release".filter_id = $1;`
+WHERE release_action_status.status = 'PUSH_APPROVED' AND release_action_status.filter_id = $1;`
 
 	row := tx.QueryRowContext(ctx, query, filterID)
 	if err := row.Err(); err != nil {

internal/database/postgres_migrate.go

@@ -264,12 +264,14 @@ CREATE TABLE release_action_status
 	type          TEXT NOT NULL,
 	client        TEXT,
 	filter        TEXT,
+	filter_id     INTEGER,
 	rejections    TEXT []   DEFAULT '{}' NOT NULL,
 	timestamp     TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 	raw           TEXT,
 	log           TEXT,
 	release_id    INTEGER NOT NULL,
-	FOREIGN KEY (release_id) REFERENCES "release"(id) ON DELETE CASCADE
+	FOREIGN KEY (release_id) REFERENCES "release"(id) ON DELETE CASCADE,
+	FOREIGN KEY (filter_id) REFERENCES "filter"(id) ON DELETE SET NULL
 );
 
 CREATE INDEX release_action_status_release_id_index
@@ -631,4 +633,18 @@ CREATE INDEX indexer_identifier_index
 	ALTER TABLE "filter"
 		ADD COLUMN except_language TEXT []   DEFAULT '{}';
 	`,
+	`ALTER TABLE release_action_status
+    ADD filter_id INTEGER;
+
+CREATE INDEX release_action_status_filter_id_index
+    ON release_action_status (filter_id);
+
+ALTER TABLE release_action_status
+    ADD CONSTRAINT release_action_status_filter_id_fk
+        FOREIGN KEY (filter_id) REFERENCES filter;
+
+UPDATE release_action_status
+SET filter_id = (SELECT f.id
+FROM filter f WHERE f.name = release_action_status.filter);
+	`,
 }

internal/database/release.go

@@ -77,8 +77,8 @@ func (repo *ReleaseRepo) StoreReleaseActionStatus(ctx context.Context, a *domain
 	} else {
 		queryBuilder := repo.db.squirrel.
 			Insert("release_action_status").
-			Columns("status", "action", "type", "client", "filter", "rejections", "timestamp", "release_id").
+			Columns("status", "action", "type", "client", "filter", "filter_id", "rejections", "timestamp", "release_id").
-			Values(a.Status, a.Action, a.Type, a.Client, a.Filter, pq.Array(a.Rejections), a.Timestamp, a.ReleaseID).
+			Values(a.Status, a.Action, a.Type, a.Client, a.Filter, a.FilterID, pq.Array(a.Rejections), a.Timestamp, a.ReleaseID).
 			Suffix("RETURNING id").RunWith(repo.db.handler)
 
 		// return values
@@ -391,7 +391,7 @@ func (repo *ReleaseRepo) GetActionStatusByReleaseID(ctx context.Context, release
 
 func (repo *ReleaseRepo) attachActionStatus(ctx context.Context, tx *Tx, releaseID int64) ([]domain.ReleaseActionStatus, error) {
 	queryBuilder := repo.db.squirrel.
-		Select("id", "status", "action", "type", "client", "filter", "rejections", "timestamp").
+		Select("id", "status", "action", "type", "client", "filter", "filter_id", "rejections", "timestamp").
 		From("release_action_status").
 		Where(sq.Eq{"release_id": releaseID})
 
@@ -417,13 +417,15 @@ func (repo *ReleaseRepo) attachActionStatus(ctx context.Context, tx *Tx, release
 		var rls domain.ReleaseActionStatus
 
 		var client, filter sql.NullString
+		var filterID sql.NullInt64
 
-		if err := rows.Scan(&rls.ID, &rls.Status, &rls.Action, &rls.Type, &client, &filter, pq.Array(&rls.Rejections), &rls.Timestamp); err != nil {
+		if err := rows.Scan(&rls.ID, &rls.Status, &rls.Action, &rls.Type, &client, &filter, &filterID, pq.Array(&rls.Rejections), &rls.Timestamp); err != nil {
 			return res, errors.Wrap(err, "error scanning row")
 		}
 
 		rls.Client = client.String
 		rls.Filter = filter.String
+		rls.FilterID = filterID.Int64
 
 		res = append(res, rls)
 	}

internal/database/sqlite_migrate.go

@@ -247,17 +247,28 @@ CREATE TABLE release_action_status
 	type          TEXT NOT NULL,
 	client        TEXT,
 	filter        TEXT,
+    filter_id     INTEGER
+        CONSTRAINT release_action_status_filter_id_fk
+            REFERENCES filter,
 	rejections    TEXT []   DEFAULT '{}' NOT NULL,
 	timestamp     TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 	raw           TEXT,
 	log           TEXT,
-	release_id    INTEGER NOT NULL,
+    release_id    INTEGER NOT NULL
-	FOREIGN KEY (release_id) REFERENCES "release"(id) ON DELETE CASCADE
+        CONSTRAINT release_action_status_release_id_fkey
+            REFERENCES "release"
+            ON DELETE CASCADE
 );
 
+CREATE INDEX release_action_status_status_index
+    ON release_action_status (status);
+
 CREATE INDEX release_action_status_release_id_index
     ON release_action_status (release_id);
 
+CREATE INDEX release_action_status_filter_id_index
+    ON release_action_status (filter_id);
+
 CREATE TABLE notification
 (
 	id         INTEGER PRIMARY KEY,
@@ -975,4 +986,58 @@ ALTER TABLE irc_network_dg_tmp
 	ALTER TABLE "filter"
 		ADD COLUMN except_language TEXT []   DEFAULT '{}';
 	`,
+	`CREATE TABLE release_action_status_dg_tmp
+(
+    id         INTEGER
+        PRIMARY KEY,
+    status     TEXT,
+    action     TEXT                   NOT NULL,
+    type       TEXT                   NOT NULL,
+    rejections TEXT      DEFAULT '{}' NOT NULL,
+    timestamp  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    raw        TEXT,
+    log        TEXT,
+    release_id INTEGER                NOT NULL
+        constraint release_action_status_release_id_fkey
+            references "release"
+            on delete cascade,
+    client     TEXT,
+    filter     TEXT,
+    filter_id  INTEGER
+        CONSTRAINT release_action_status_filter_id_fk
+            REFERENCES filter
+);
+
+INSERT INTO release_action_status_dg_tmp(id, status, action, type, rejections, timestamp, raw, log, release_id, client, filter)
+SELECT id,
+       status,
+       action,
+       type,
+       rejections,
+       timestamp,
+       raw,
+       log,
+       release_id,
+       client,
+       filter
+FROM release_action_status;
+
+DROP TABLE release_action_status;
+
+ALTER TABLE release_action_status_dg_tmp
+    RENAME TO release_action_status;
+
+CREATE INDEX release_action_status_filter_id_index
+    ON release_action_status (filter_id);
+
+CREATE INDEX release_action_status_release_id_index
+    ON release_action_status (release_id);
+
+CREATE INDEX release_action_status_status_index
+    ON release_action_status (status);
+
+UPDATE release_action_status
+SET filter_id = (SELECT f.id
+FROM filter f WHERE f.name = release_action_status.filter);
+	`,
 }

internal/domain/release.go

@@ -98,6 +98,7 @@ type ReleaseActionStatus struct {
 	Type       ActionType        `json:"type"`
 	Client     string            `json:"client"`
 	Filter     string            `json:"filter"`
+	FilterID   int64             `json:"-"`
 	Rejections []string          `json:"rejections"`
 	Timestamp  time.Time         `json:"timestamp"`
 	ReleaseID  int64             `json:"-"`