feat(releases): replay actions (#932)

* feat(releases): replay actions

* feat(releases): replay actions component

* fix: update filter actions

* fix: select filter_id from ras
This commit is contained in:
ze0s 2023-05-15 21:30:04 +02:00 committed by GitHub
parent 97333d334f
commit 6898ad8315
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 752 additions and 189 deletions

View file

@ -280,30 +280,117 @@ func (r *ActionRepo) List(ctx context.Context) ([]domain.Action, error) {
a.ClientID = clientID.Int32
actions = append(actions, a)
}
if err := rows.Err(); err != nil {
return nil, errors.Wrap(err, "rows error")
if err := rows.Err(); err != nil {
return nil, errors.Wrap(err, "rows error")
}
}
return actions, nil
}
func (r *ActionRepo) Delete(actionID int) error {
func (r *ActionRepo) Get(ctx context.Context, req *domain.GetActionRequest) (*domain.Action, error) {
queryBuilder := r.db.squirrel.
Select(
"id",
"name",
"type",
"enabled",
"exec_cmd",
"exec_args",
"watch_folder",
"category",
"tags",
"label",
"save_path",
"paused",
"ignore_rules",
"limit_download_speed",
"limit_upload_speed",
"limit_ratio",
"limit_seed_time",
"reannounce_skip",
"reannounce_delete",
"reannounce_interval",
"reannounce_max_attempts",
"webhook_host",
"webhook_type",
"webhook_method",
"webhook_data",
"client_id",
"filter_id",
).
From("action").
Where(sq.Eq{"id": req.Id})
query, args, err := queryBuilder.ToSql()
if err != nil {
return nil, errors.Wrap(err, "error building query")
}
row := r.db.handler.QueryRowContext(ctx, query, args...)
if err != nil {
return nil, errors.Wrap(err, "error executing query")
}
if err := row.Err(); err != nil {
return nil, errors.Wrap(err, "rows error")
}
var a domain.Action
var execCmd, execArgs, watchFolder, category, tags, label, savePath, webhookHost, webhookType, webhookMethod, webhookData sql.NullString
var limitUl, limitDl, limitSeedTime sql.NullInt64
var limitRatio sql.NullFloat64
var clientID, filterID sql.NullInt32
var paused, ignoreRules sql.NullBool
if err := row.Scan(&a.ID, &a.Name, &a.Type, &a.Enabled, &execCmd, &execArgs, &watchFolder, &category, &tags, &label, &savePath, &paused, &ignoreRules, &limitDl, &limitUl, &limitRatio, &limitSeedTime, &a.ReAnnounceSkip, &a.ReAnnounceDelete, &a.ReAnnounceInterval, &a.ReAnnounceMaxAttempts, &webhookHost, &webhookType, &webhookMethod, &webhookData, &clientID, &filterID); err != nil {
if err == sql.ErrNoRows {
return nil, nil
}
return nil, errors.Wrap(err, "error scanning row")
}
a.Category = category.String
a.Tags = tags.String
a.Label = label.String
a.SavePath = savePath.String
a.Paused = paused.Bool
a.IgnoreRules = ignoreRules.Bool
a.LimitDownloadSpeed = limitDl.Int64
a.LimitUploadSpeed = limitUl.Int64
a.LimitRatio = limitRatio.Float64
a.LimitSeedTime = limitSeedTime.Int64
a.WebhookHost = webhookHost.String
a.WebhookType = webhookType.String
a.WebhookMethod = webhookMethod.String
a.WebhookData = webhookData.String
a.ClientID = clientID.Int32
a.FilterID = int(filterID.Int32)
return &a, nil
}
func (r *ActionRepo) Delete(ctx context.Context, req *domain.DeleteActionRequest) error {
queryBuilder := r.db.squirrel.
Delete("action").
Where(sq.Eq{"id": actionID})
Where(sq.Eq{"id": req.ActionId})
query, args, err := queryBuilder.ToSql()
if err != nil {
return errors.Wrap(err, "error building query")
}
_, err = r.db.handler.Exec(query, args...)
if err != nil {
if _, err = r.db.handler.ExecContext(ctx, query, args...); err != nil {
return errors.Wrap(err, "error executing query")
}
r.log.Debug().Msgf("action.delete: %v", actionID)
r.log.Debug().Msgf("action.delete: %v", req.ActionId)
return nil
}
@ -506,113 +593,171 @@ func (r *ActionRepo) StoreFilterActions(ctx context.Context, actions []*domain.A
defer tx.Rollback()
deleteQueryBuilder := r.db.squirrel.
Delete("action").
Where(sq.Eq{"filter_id": filterID})
deleteQuery, deleteArgs, err := deleteQueryBuilder.ToSql()
if err != nil {
return nil, errors.Wrap(err, "error building query")
}
_, err = tx.ExecContext(ctx, deleteQuery, deleteArgs...)
if err != nil {
return nil, errors.Wrap(err, "error executing query")
}
for _, action := range actions {
execCmd := toNullString(action.ExecCmd)
execArgs := toNullString(action.ExecArgs)
watchFolder := toNullString(action.WatchFolder)
category := toNullString(action.Category)
tags := toNullString(action.Tags)
label := toNullString(action.Label)
savePath := toNullString(action.SavePath)
contentLayout := toNullString(string(action.ContentLayout))
webhookHost := toNullString(action.WebhookHost)
webhookType := toNullString(action.WebhookType)
webhookMethod := toNullString(action.WebhookMethod)
webhookData := toNullString(action.WebhookData)
action := action
limitDL := toNullInt64(action.LimitDownloadSpeed)
limitUL := toNullInt64(action.LimitUploadSpeed)
limitRatio := toNullFloat64(action.LimitRatio)
limitSeedTime := toNullInt64(action.LimitSeedTime)
clientID := toNullInt32(action.ClientID)
if action.ID > 0 {
execCmd := toNullString(action.ExecCmd)
execArgs := toNullString(action.ExecArgs)
watchFolder := toNullString(action.WatchFolder)
category := toNullString(action.Category)
tags := toNullString(action.Tags)
label := toNullString(action.Label)
savePath := toNullString(action.SavePath)
contentLayout := toNullString(string(action.ContentLayout))
webhookHost := toNullString(action.WebhookHost)
webhookType := toNullString(action.WebhookType)
webhookMethod := toNullString(action.WebhookMethod)
webhookData := toNullString(action.WebhookData)
queryBuilder := r.db.squirrel.
Insert("action").
Columns(
"name",
"type",
"enabled",
"exec_cmd",
"exec_args",
"watch_folder",
"category",
"tags",
"label",
"save_path",
"paused",
"ignore_rules",
"skip_hash_check",
"content_layout",
"limit_upload_speed",
"limit_download_speed",
"limit_ratio",
"limit_seed_time",
"reannounce_skip",
"reannounce_delete",
"reannounce_interval",
"reannounce_max_attempts",
"webhook_host",
"webhook_type",
"webhook_method",
"webhook_data",
"client_id",
"filter_id",
).
Values(
action.Name,
action.Type,
action.Enabled,
execCmd,
execArgs,
watchFolder,
category,
tags,
label,
savePath,
action.Paused,
action.IgnoreRules,
action.SkipHashCheck,
contentLayout,
limitUL,
limitDL,
limitRatio,
limitSeedTime,
action.ReAnnounceSkip,
action.ReAnnounceDelete,
action.ReAnnounceInterval,
action.ReAnnounceMaxAttempts,
webhookHost,
webhookType,
webhookMethod,
webhookData,
clientID,
filterID,
).
Suffix("RETURNING id").RunWith(tx)
limitDL := toNullInt64(action.LimitDownloadSpeed)
limitUL := toNullInt64(action.LimitUploadSpeed)
limitRatio := toNullFloat64(action.LimitRatio)
limitSeedTime := toNullInt64(action.LimitSeedTime)
// return values
var retID int
clientID := toNullInt32(action.ClientID)
err = queryBuilder.QueryRowContext(ctx).Scan(&retID)
if err != nil {
return nil, errors.Wrap(err, "error executing query")
var err error
queryBuilder := r.db.squirrel.
Update("action").
Set("name", action.Name).
Set("type", action.Type).
Set("enabled", action.Enabled).
Set("exec_cmd", execCmd).
Set("exec_args", execArgs).
Set("watch_folder", watchFolder).
Set("category", category).
Set("tags", tags).
Set("label", label).
Set("save_path", savePath).
Set("paused", action.Paused).
Set("ignore_rules", action.IgnoreRules).
Set("skip_hash_check", action.SkipHashCheck).
Set("content_layout", contentLayout).
Set("limit_upload_speed", limitUL).
Set("limit_download_speed", limitDL).
Set("limit_ratio", limitRatio).
Set("limit_seed_time", limitSeedTime).
Set("reannounce_skip", action.ReAnnounceSkip).
Set("reannounce_delete", action.ReAnnounceDelete).
Set("reannounce_interval", action.ReAnnounceInterval).
Set("reannounce_max_attempts", action.ReAnnounceMaxAttempts).
Set("webhook_host", webhookHost).
Set("webhook_type", webhookType).
Set("webhook_method", webhookMethod).
Set("webhook_data", webhookData).
Set("client_id", clientID).
Set("filter_id", filterID).
Where(sq.Eq{"id": action.ID})
query, args, err := queryBuilder.ToSql()
if err != nil {
return nil, errors.Wrap(err, "error building query")
}
if _, err = tx.ExecContext(ctx, query, args...); err != nil {
return nil, errors.Wrap(err, "error executing query")
}
r.log.Trace().Msgf("action.StoreFilterActions: update %d", action.ID)
} else {
execCmd := toNullString(action.ExecCmd)
execArgs := toNullString(action.ExecArgs)
watchFolder := toNullString(action.WatchFolder)
category := toNullString(action.Category)
tags := toNullString(action.Tags)
label := toNullString(action.Label)
savePath := toNullString(action.SavePath)
contentLayout := toNullString(string(action.ContentLayout))
webhookHost := toNullString(action.WebhookHost)
webhookType := toNullString(action.WebhookType)
webhookMethod := toNullString(action.WebhookMethod)
webhookData := toNullString(action.WebhookData)
limitDL := toNullInt64(action.LimitDownloadSpeed)
limitUL := toNullInt64(action.LimitUploadSpeed)
limitRatio := toNullFloat64(action.LimitRatio)
limitSeedTime := toNullInt64(action.LimitSeedTime)
clientID := toNullInt32(action.ClientID)
queryBuilder := r.db.squirrel.
Insert("action").
Columns(
"name",
"type",
"enabled",
"exec_cmd",
"exec_args",
"watch_folder",
"category",
"tags",
"label",
"save_path",
"paused",
"ignore_rules",
"skip_hash_check",
"content_layout",
"limit_upload_speed",
"limit_download_speed",
"limit_ratio",
"limit_seed_time",
"reannounce_skip",
"reannounce_delete",
"reannounce_interval",
"reannounce_max_attempts",
"webhook_host",
"webhook_type",
"webhook_method",
"webhook_data",
"client_id",
"filter_id",
).
Values(
action.Name,
action.Type,
action.Enabled,
execCmd,
execArgs,
watchFolder,
category,
tags,
label,
savePath,
action.Paused,
action.IgnoreRules,
action.SkipHashCheck,
contentLayout,
limitUL,
limitDL,
limitRatio,
limitSeedTime,
action.ReAnnounceSkip,
action.ReAnnounceDelete,
action.ReAnnounceInterval,
action.ReAnnounceMaxAttempts,
webhookHost,
webhookType,
webhookMethod,
webhookData,
clientID,
filterID,
).
Suffix("RETURNING id").RunWith(tx)
// return values
var retID int
if err = queryBuilder.QueryRowContext(ctx).Scan(&retID); err != nil {
return nil, errors.Wrap(err, "error executing query")
}
action.ID = retID
r.log.Trace().Msgf("action.StoreFilterActions: store %d", action.ID)
}
action.ID = retID
r.log.Debug().Msgf("action.StoreFilterActions: store '%v' type: '%v' on filter: %v", action.Name, action.Type, filterID)
}

View file

@ -271,6 +271,7 @@ CREATE TABLE release_action_status
id SERIAL PRIMARY KEY,
status TEXT,
action TEXT NOT NULL,
action_id INTEGER,
type TEXT NOT NULL,
client TEXT,
filter TEXT,
@ -280,6 +281,7 @@ CREATE TABLE release_action_status
raw TEXT,
log TEXT,
release_id INTEGER NOT NULL,
FOREIGN KEY (action_id) REFERENCES "action"(id),
FOREIGN KEY (release_id) REFERENCES "release"(id) ON DELETE CASCADE,
FOREIGN KEY (filter_id) REFERENCES "filter"(id) ON DELETE SET NULL
);
@ -691,4 +693,10 @@ ADD COLUMN topic text;`,
ALTER TABLE filter
ADD COLUMN use_regex_description BOOLEAN DEFAULT FALSE;`,
`ALTER TABLE release_action_status
ADD action_id INTEGER;
ALTER TABLE release_action_status
ADD CONSTRAINT release_action_status_action_id_fk
FOREIGN KEY (action_id) REFERENCES action;`,
}

View file

@ -79,8 +79,8 @@ func (repo *ReleaseRepo) StoreReleaseActionStatus(ctx context.Context, status *d
} else {
queryBuilder := repo.db.squirrel.
Insert("release_action_status").
Columns("status", "action", "type", "client", "filter", "filter_id", "rejections", "timestamp", "release_id").
Values(status.Status, status.Action, status.Type, status.Client, status.Filter, status.FilterID, pq.Array(status.Rejections), status.Timestamp.Format(time.RFC3339), status.ReleaseID).
Columns("status", "action", "action_id", "type", "client", "filter", "filter_id", "rejections", "timestamp", "release_id").
Values(status.Status, status.Action, status.ActionID, status.Type, status.Client, status.Filter, status.FilterID, pq.Array(status.Rejections), status.Timestamp.Format(time.RFC3339), status.ReleaseID).
Suffix("RETURNING id").RunWith(repo.db.handler)
// return values
@ -207,7 +207,7 @@ func (repo *ReleaseRepo) findReleases(ctx context.Context, tx *Tx, params domain
queryBuilder := repo.db.squirrel.
Select("r.id", "r.filter_status", "r.rejections", "r.indexer", "r.filter", "r.protocol", "r.info_url", "r.download_url", "r.title", "r.torrent_name", "r.size", "r.timestamp",
"ras.id", "ras.status", "ras.action", "ras.type", "ras.client", "ras.filter", "ras.rejections", "ras.timestamp").
"ras.id", "ras.status", "ras.action", "ras.action_id", "ras.type", "ras.client", "ras.filter", "ras.filter_id", "ras.release_id", "ras.rejections", "ras.timestamp").
Column(sq.Alias(countQuery, "page_total")).
From("release r").
OrderBy("r.id DESC").
@ -242,22 +242,25 @@ func (repo *ReleaseRepo) findReleases(ctx context.Context, tx *Tx, params domain
var rlsindexer, rlsfilter, infoUrl, downloadUrl sql.NullString
var rasId sql.NullInt64
var rasId, rasFilterId, rasReleaseId, rasActionId sql.NullInt64
var rasStatus, rasAction, rasType, rasClient, rasFilter sql.NullString
var rasRejections []sql.NullString
var rasTimestamp sql.NullTime
if err := rows.Scan(&rls.ID, &rls.FilterStatus, pq.Array(&rls.Rejections), &rlsindexer, &rlsfilter, &rls.Protocol, &infoUrl, &downloadUrl, &rls.Title, &rls.TorrentName, &rls.Size, &rls.Timestamp, &rasId, &rasStatus, &rasAction, &rasType, &rasClient, &rasFilter, pq.Array(&rasRejections), &rasTimestamp, &countItems); err != nil {
if err := rows.Scan(&rls.ID, &rls.FilterStatus, pq.Array(&rls.Rejections), &rlsindexer, &rlsfilter, &rls.Protocol, &infoUrl, &downloadUrl, &rls.Title, &rls.TorrentName, &rls.Size, &rls.Timestamp, &rasId, &rasStatus, &rasAction, &rasActionId, &rasType, &rasClient, &rasFilter, &rasFilterId, &rasReleaseId, pq.Array(&rasRejections), &rasTimestamp, &countItems); err != nil {
return res, 0, 0, errors.Wrap(err, "error scanning row")
}
ras.ID = rasId.Int64
ras.Status = domain.ReleasePushStatus(rasStatus.String)
ras.Action = rasAction.String
ras.ActionID = rasActionId.Int64
ras.Type = domain.ActionType(rasType.String)
ras.Client = rasClient.String
ras.Filter = rasFilter.String
ras.FilterID = rasFilterId.Int64
ras.Timestamp = rasTimestamp.Time
ras.ReleaseID = rasReleaseId.Int64
ras.Rejections = []string{}
for _, rejection := range rasRejections {
@ -351,7 +354,7 @@ func (repo *ReleaseRepo) GetIndexerOptions(ctx context.Context) ([]string, error
func (repo *ReleaseRepo) GetActionStatusByReleaseID(ctx context.Context, releaseID int64) ([]domain.ReleaseActionStatus, error) {
queryBuilder := repo.db.squirrel.
Select("id", "status", "action", "type", "client", "filter", "rejections", "timestamp").
Select("id", "status", "action", "action_id", "type", "client", "filter", "release_id", "rejections", "timestamp").
From("release_action_status").
Where(sq.Eq{"release_id": releaseID})
@ -378,11 +381,13 @@ func (repo *ReleaseRepo) GetActionStatusByReleaseID(ctx context.Context, release
var rls domain.ReleaseActionStatus
var client, filter sql.NullString
var actionId 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, &actionId, &rls.Type, &client, &filter, &rls.ReleaseID, pq.Array(&rls.Rejections), &rls.Timestamp); err != nil {
return res, errors.Wrap(err, "error scanning row")
}
rls.ActionID = actionId.Int64
rls.Client = client.String
rls.Filter = filter.String
@ -392,9 +397,96 @@ func (repo *ReleaseRepo) GetActionStatusByReleaseID(ctx context.Context, release
return res, nil
}
func (repo *ReleaseRepo) Get(ctx context.Context, req *domain.GetReleaseRequest) (*domain.Release, error) {
queryBuilder := repo.db.squirrel.
Select("r.id", "r.filter_status", "r.rejections", "r.indexer", "r.filter", "r.filter_id", "r.protocol", "r.info_url", "r.download_url", "r.title", "r.torrent_name", "r.size", "r.timestamp").
From("release r").
OrderBy("r.id DESC").
Where(sq.Eq{"r.id": req.Id})
query, args, err := queryBuilder.ToSql()
if err != nil {
return nil, errors.Wrap(err, "error building query")
}
repo.log.Trace().Str("database", "release.find").Msgf("query: '%v', args: '%v'", query, args)
row := repo.db.handler.QueryRowContext(ctx, query, args...)
if err != nil {
return nil, errors.Wrap(err, "error executing query")
}
if err := row.Err(); err != nil {
return nil, errors.Wrap(err, "error rows find release")
}
var rls domain.Release
var indexerName, filterName, infoUrl, downloadUrl sql.NullString
var filterId sql.NullInt64
if err := row.Scan(&rls.ID, &rls.FilterStatus, pq.Array(&rls.Rejections), &indexerName, &filterName, &filterId, &rls.Protocol, &infoUrl, &downloadUrl, &rls.Title, &rls.TorrentName, &rls.Size, &rls.Timestamp); err != nil {
if err == sql.ErrNoRows {
return nil, nil
}
return nil, errors.Wrap(err, "error scanning row")
}
rls.Indexer = indexerName.String
rls.FilterName = filterName.String
rls.FilterID = int(filterId.Int64)
rls.ActionStatus = make([]domain.ReleaseActionStatus, 0)
rls.InfoURL = infoUrl.String
rls.TorrentURL = downloadUrl.String
return &rls, nil
}
func (repo *ReleaseRepo) GetActionStatus(ctx context.Context, req *domain.GetReleaseActionStatusRequest) (*domain.ReleaseActionStatus, error) {
queryBuilder := repo.db.squirrel.
Select("id", "status", "action", "action_id", "type", "client", "filter", "filter_id", "release_id", "rejections", "timestamp").
From("release_action_status").
Where(sq.Eq{"id": req.Id})
query, args, err := queryBuilder.ToSql()
if err != nil {
return nil, errors.Wrap(err, "error building query")
}
row := repo.db.handler.QueryRowContext(ctx, query, args...)
if err != nil {
return nil, errors.Wrap(err, "error executing query")
}
if err := row.Err(); err != nil {
repo.log.Error().Stack().Err(err)
return nil, err
}
var rls domain.ReleaseActionStatus
var client, filter sql.NullString
var actionId, filterId sql.NullInt64
if err := row.Scan(&rls.ID, &rls.Status, &rls.Action, &actionId, &rls.Type, &client, &filter, &filterId, &rls.ReleaseID, pq.Array(&rls.Rejections), &rls.Timestamp); err != nil {
if err == sql.ErrNoRows {
return nil, nil
}
return nil, errors.Wrap(err, "error scanning row")
}
rls.ActionID = actionId.Int64
rls.Client = client.String
rls.Filter = filter.String
rls.FilterID = filterId.Int64
return &rls, nil
}
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", "filter_id", "rejections", "timestamp").
Select("id", "status", "action", "action_id", "type", "client", "filter", "filter_id", "release_id", "rejections", "timestamp").
From("release_action_status").
Where(sq.Eq{"release_id": releaseID})
@ -420,12 +512,13 @@ func (repo *ReleaseRepo) attachActionStatus(ctx context.Context, tx *Tx, release
var rls domain.ReleaseActionStatus
var client, filter sql.NullString
var filterID sql.NullInt64
var actionId, filterID sql.NullInt64
if err := rows.Scan(&rls.ID, &rls.Status, &rls.Action, &rls.Type, &client, &filter, &filterID, pq.Array(&rls.Rejections), &rls.Timestamp); err != nil {
if err := rows.Scan(&rls.ID, &rls.Status, &rls.Action, &actionId, &rls.Type, &client, &filter, &filterID, &rls.ReleaseID, pq.Array(&rls.Rejections), &rls.Timestamp); err != nil {
return res, errors.Wrap(err, "error scanning row")
}
rls.ActionID = actionId.Int64
rls.Client = client.String
rls.Filter = filter.String
rls.FilterID = filterID.Int64

View file

@ -254,6 +254,9 @@ CREATE TABLE release_action_status
id INTEGER PRIMARY KEY,
status TEXT,
action TEXT NOT NULL,
action_id INTEGER
CONSTRAINT release_action_status_action_id_fk
REFERENCES action,
type TEXT NOT NULL,
client TEXT,
filter TEXT,
@ -1084,4 +1087,58 @@ ADD COLUMN topic text;`,
ALTER TABLE filter
ADD COLUMN use_regex_description BOOLEAN DEFAULT FALSE;`,
`create table release_action_status_dg_tmp
(
id INTEGER
primary key,
status TEXT,
action TEXT not null,
action_id INTEGER
constraint release_action_status_action_id_fk
references action,
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, filter_id)
select id,
status,
action,
type,
rejections,
timestamp,
raw,
log,
release_id,
client,
filter,
filter_id
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);`,
}