package server import ( "errors" "git.myservermanager.com/varakh/upda/api" "git.myservermanager.com/varakh/upda/util" "github.com/containrrr/shoutrrr" "go.uber.org/zap" "strings" "time" ) type actionInvocationService struct { repo ActionInvocationRepository actionService *actionService eventService *eventService secretService *secretService } func newActionInvocationService(r ActionInvocationRepository, a *actionService, e *eventService, s *secretService) *actionInvocationService { return &actionInvocationService{ repo: r, actionService: a, eventService: e, secretService: s, } } func (s *actionInvocationService) enqueue(batchSize int) error { if batchSize <= 0 { return newServiceError(General, errors.New("cannot enqueue actions from events with invalid configured batch size")) } var events []*Event var err error if events, err = s.eventService.getByState(batchSize, api.EventStateCreated); err != nil { return err } var actions []*Action if actions, err = s.actionService.getAll(); err != nil { return err } for _, event := range events { if err = s.enqueueFromEvent(event, actions); err != nil { zap.L().Sugar().Errorf("Could not enqueue action for event '%s' (%s). Reason: %s", event.Name, event.ID, err.Error()) } } return nil } func (s *actionInvocationService) enqueueFromEvent(event *Event, actions []*Action) error { if event == nil || actions == nil { return newServiceError(IllegalArgument, errorValidationNotEmpty) } var err error // match requires event payload var eventPayload *eventPayloadInformationDto if eventPayload, err = s.eventService.extractPayloadInfo(event); err != nil { return err } var filteredActions []*Action for _, action := range actions { matchesEvent := action.MatchEvent == nil || *action.MatchEvent == event.Name matchesHost := action.MatchHost == nil || *action.MatchHost == eventPayload.Host matchesApplication := action.MatchApplication == nil || *action.MatchApplication == eventPayload.Application matchesProvider := action.MatchProvider == nil || *action.MatchProvider == eventPayload.Provider if matchesEvent && matchesHost && matchesApplication && matchesProvider { filteredActions = append(filteredActions, action) } } if len(filteredActions) == 0 { zap.L().Sugar().Debugf("No actions found which match event '%s', nothing to enqueue", event.Name) return nil } for _, action := range filteredActions { if _, err = s.create(event, action, api.ActionInvocationStateCreated); err != nil { zap.L().Sugar().Errorf("Could not enqueue action '%s' (%v). Reason: %s", action.Label, action.ID, err.Error()) continue } } // mark event as enqueued if _, err = s.eventService.updateState(event.ID.String(), api.EventStateEnqueued); err != nil { return err } return nil } func (s *actionInvocationService) invoke(batchSize int, maxRetries int) error { if batchSize <= 0 { return newServiceError(General, errors.New("cannot invoke actions with invalid configured batch size")) } if maxRetries <= 0 { return newServiceError(General, errors.New("cannot invoke actions with invalid configured max retries")) } var err error var actionInvocations []*ActionInvocation if actionInvocations, err = s.getByState(batchSize, maxRetries, api.ActionInvocationStateCreated, api.ActionInvocationStateError); err != nil { return err } if len(actionInvocations) == 0 { zap.L().Sugar().Debugf("No action invocations found to process") return nil } for _, actionInvocation := range actionInvocations { if _, err = s.updateState(actionInvocation.ID.String(), api.ActionInvocationStateRunning); err != nil { zap.L().Sugar().Errorf("Could not mark action invocation '%v' as running. Reason: %s", actionInvocation.ID, err.Error()) continue } zap.L().Sugar().Debugf("Invoking action '%v' for event '%v'", actionInvocation.ActionID, actionInvocation.EventID) var event *Event if event, err = s.eventService.get(actionInvocation.EventID); err != nil { zap.L().Sugar().Errorf("Could not find event '%v' for action '%v' and action invocation '%v'. Reason: %s", actionInvocation.EventID, actionInvocation.ActionID, actionInvocation.ID, err.Error()) // with cascade, cannot happen continue } var eventPayload *eventPayloadInformationDto if eventPayload, err = s.eventService.extractPayloadInfo(event); err != nil { zap.L().Sugar().Errorf("Could not extract event's '%v' information for action '%v' and action invocation '%v'. Reason: %s", actionInvocation.EventID, actionInvocation.ActionID, actionInvocation.ID, err.Error()) // with layout of attached payload, cannot happen continue } var action *Action if action, err = s.actionService.get(actionInvocation.ActionID); err != nil { zap.L().Sugar().Errorf("Could not find action '%v' for action invocation '%v'. Reason: %s", actionInvocation.ActionID, actionInvocation.ID, err.Error()) // with cascade, cannot happen continue } if err = s.execute(action, eventPayload); err != nil { var cause error cause = err zap.L().Sugar().Errorf("Could not invoke action '%s' (%v) for action invocation '%v'. Reason: %s", action.Label, action.ID, actionInvocation.ID, err.Error()) if _, err = s.updateState(actionInvocation.ID.String(), api.ActionInvocationStateError); err != nil { zap.L().Sugar().Errorf("Could not mark action invocation '%v' as error. Reason: %s", actionInvocation.ID, err.Error()) } msg := cause.Error() if _, err = s.updateMessage(actionInvocation.ID.String(), &msg); err != nil { zap.L().Sugar().Errorf("Could not update action invocation '%v' message. Reason: %s", actionInvocation.ID, err.Error()) } newRetryCount := actionInvocation.RetryCount + 1 if _, err = s.updateRetryCount(actionInvocation.ID.String(), newRetryCount); err != nil { zap.L().Sugar().Errorf("Could not update action invocation '%v' retry count to '%d'. Reason: %s", actionInvocation.ID, newRetryCount, err.Error()) } if newRetryCount >= maxRetries { zap.L().Sugar().Infof("Action invocation '%v' exceeded max retry count of '%d'. Not trying again.", actionInvocation.ID, newRetryCount) } continue } zap.L().Sugar().Debugf("Processed action invocation '%v' for event '%s' (%v) and action '%s' (%v)", actionInvocation.ID, event.Name, event.ID, action.Label, action.ID) if _, err = s.updateState(actionInvocation.ID.String(), api.ActionInvocationStateSuccess); err != nil { zap.L().Sugar().Errorf("Could not mark action invocation '%v' as success. Reason: %s", actionInvocation.ID, err.Error()) } if _, err = s.updateMessage(actionInvocation.ID.String(), nil); err != nil { zap.L().Sugar().Errorf("Could not update action invocation '%v' message. Reason: %s", actionInvocation.ID, err.Error()) } } return nil } func (s *actionInvocationService) execute(action *Action, eventPayloadInfo *eventPayloadInformationDto) error { if action == nil || eventPayloadInfo == nil { return errorValidationNotEmpty } var err error var bytes []byte if bytes, err = action.Payload.MarshalJSON(); err != nil { return newServiceError(General, err) } switch action.Type { case api.ActionTypeShoutrrr.Value(): var payload actionPayloadShoutrrrDto if payload, err = util.UnmarshalGenericJSON[actionPayloadShoutrrrDto](bytes); err != nil { return newServiceError(General, err) } body := s.replaceVars(payload.Body, eventPayloadInfo) body = s.replaceSecrets(body) for _, url := range payload.Urls { url = s.replaceSecrets(url) url = s.replaceVars(url, eventPayloadInfo) if err = shoutrrr.Send(url, body); err != nil { return err } } break default: return newServiceError(General, errors.New("no matching action type found for invocation")) } return nil } func (s *actionInvocationService) replaceSecrets(str string) string { if str == "" { return str } var matches [][]string matches = util.ExtractBetween(str, "", "") var err error for _, match := range matches { var val string if val, err = s.secretService.getValueByKey(match[1]); err != nil { zap.L().Sugar().Warnf("Could not inject secret '%s'. Reason: %s", match[1], err.Error()) continue } str = strings.ReplaceAll(str, match[0], val) } return str } func (s *actionInvocationService) replaceVars(str string, eventPayloadInfo *eventPayloadInformationDto) string { if str == "" || eventPayloadInfo == nil { return str } str = strings.ReplaceAll(str, "APPLICATION", eventPayloadInfo.Application) str = strings.ReplaceAll(str, "PROVIDER", eventPayloadInfo.Provider) str = strings.ReplaceAll(str, "HOST", eventPayloadInfo.Host) str = strings.ReplaceAll(str, "VERSION", eventPayloadInfo.Version) return str } func (s *actionInvocationService) paginate(page int, pageSize int, orderBy string, order string) ([]*ActionInvocation, error) { return s.repo.paginate(page, pageSize, orderBy, order) } func (s *actionInvocationService) get(id string) (*ActionInvocation, error) { if id == "" { return nil, errorValidationNotBlank } e, err := s.repo.find(id) if err != nil { return nil, err } return e, nil } func (s *actionInvocationService) getByState(limit int, maxRetries int, state ...api.ActionInvocationState) ([]*ActionInvocation, error) { if len(state) == 0 { return nil, errorValidationNotEmpty } if limit <= 0 { return nil, errorValidationLimitGreaterZero } if maxRetries <= 0 { return nil, errorValidationMaxRetriesGreaterZero } return s.repo.findAllByState(limit, maxRetries, state...) } func (s *actionInvocationService) count() (int64, error) { return s.repo.count() } func (s *actionInvocationService) delete(id string) error { if id == "" { return errorValidationNotBlank } e, err := s.get(id) if err != nil { return err } if _, err = s.repo.delete(e.ID.String()); err != nil { return err } zap.L().Sugar().Infof("Deleted action '%v'", id) return nil } func (s *actionInvocationService) updateState(id string, state api.ActionInvocationState) (*ActionInvocation, error) { if id == "" || state == "" { return nil, errorValidationNotBlank } var e *ActionInvocation var err error if e, err = s.get(id); err != nil { return nil, err } if e, err = s.repo.updateState(id, state); err != nil { return nil, err } zap.L().Sugar().Infof("Modified action invocation '%v'", id) return e, nil } func (s *actionInvocationService) updateMessage(id string, message *string) (*ActionInvocation, error) { if id == "" { return nil, errorValidationNotBlank } var e *ActionInvocation var err error if e, err = s.get(id); err != nil { return nil, err } if e, err = s.repo.updateMessage(id, message); err != nil { return nil, err } zap.L().Sugar().Infof("Modified action invocation '%v'", id) return e, nil } func (s *actionInvocationService) updateRetryCount(id string, retryCount int) (*ActionInvocation, error) { if id == "" { return nil, errorValidationNotBlank } var e *ActionInvocation var err error if e, err = s.get(id); err != nil { return nil, err } if e, err = s.repo.updateRetryCount(id, retryCount); err != nil { return nil, err } zap.L().Sugar().Infof("Modified action invocation '%v'", id) return e, nil } func (s *actionInvocationService) create(event *Event, action *Action, state api.ActionInvocationState) (*ActionInvocation, error) { if state == "" { return nil, errorValidationNotBlank } if action == nil || event == nil { return nil, errorValidationNotEmpty } var err error var e *ActionInvocation if e, err = s.repo.create(event.ID.String(), action.ID.String(), state); err != nil { return nil, err } else { zap.L().Sugar().Info("Created action invocation") return e, nil } } func (s *actionInvocationService) cleanStale(time time.Time, maxRetries int, state ...api.ActionInvocationState) (int64, error) { if len(state) == 0 { return 0, errorValidationNotEmpty } return s.repo.deleteByUpdatedAtBeforeAndStates(time, maxRetries, state...) }