diff --git a/compose/service/record.go b/compose/service/record.go index fa2133934..9d47e5aeb 100644 --- a/compose/service/record.go +++ b/compose/service/record.go @@ -82,6 +82,8 @@ type ( DeleteByID(namespaceID, moduleID uint64, recordID ...uint64) error Organize(namespaceID, moduleID, recordID uint64, sortingField, sortingValue, sortingFilter, valueField, value string) error + + Iterator(f types.RecordFilter, fn eventbus.HandlerFn, action string) (err error) } Encoder interface { @@ -795,6 +797,123 @@ func (svc record) Organize(namespaceID, moduleID, recordID uint64, posField, pos }) } +// Iterator loads and iterates through list of records +// +// For each record, RecordOnIteration is generated and passed to fn() +// to be then passed to automation script that invoked the iteration +// +// No other triggers (before/after update/delete/create) are fired when (if) +// records are changed +// +// action arg enables one of the following scenarios: +// - clone: make new record (unless aborted) +// - update: update records (unless aborted) +// - delete: delete records (unless aborted) +// - default: only iterates over records, records are not changed, return value is ignored +// +// +// Iterator can be invoked only when defined in corredor script: +// +// return default { +// iterator (each) { +// return each({ +// resourceType: 'compose:record', +// // action: 'update', +// filter: { +// namespace: '122709101053521922', +// module: '122709116471783426', +// query: 'Status = "foo"', +// sort: 'Status DESC', +// limit: 3, +// }, +// }) +// }, +// +// // this is required in case of a deferred iterator +// // security: { runAs: .... } } +// +// // exec gets called for every record found by iterator +// exec () { ... } +// } +func (svc record) Iterator(f types.RecordFilter, fn eventbus.HandlerFn, action string) (err error) { + var ( + invokerID = auth.GetIdentityFromContext(svc.ctx).Identity() + + ns *types.Namespace + m *types.Module + set types.RecordSet + ) + + return svc.db.Transaction(func() (err error) { + if ns, m, _, err = svc.loadCombo(f.NamespaceID, f.ModuleID, 0); err != nil { + return + } + + if !svc.ac.CanUpdateRecord(svc.ctx, m) { + return ErrNoUpdatePermissions.withStack() + } + + // @todo might be good to split set into smaller chunks + set, f, err = svc.recordRepo.Find(m, f) + if err != nil { + return + } + + if err = svc.preloadValues(m, set...); err != nil { + return + } + + for _, rec := range set { + if err = fn(svc.ctx, event.RecordOnIteration(rec, nil, m, ns, nil)); err != nil { + if err.Error() != "Aborted" { + // When script was softly aborted (return false), + // proceed with iteration but do not clone, update or delete + // current record! + return + } + } + + switch action { + case "clone": + var cln *types.Record + + // Assign defaults (only on missing values) + rec.Values = svc.setDefaultValues(m, rec.Values) + + // Handle payload from automation scripts + if rve := svc.procCreate(invokerID, m, rec); !rve.IsValid() { + return rve + } + + if cln, err = svc.recordRepo.Create(rec); err != nil { + return + } else if err = svc.recordRepo.UpdateValues(cln.ID, cln.Values); err != nil { + return + } + case "update": + // Handle input payload + if rve := svc.procUpdate(invokerID, m, rec, rec); !rve.IsValid() { + return rve + } + + if rec, err = svc.recordRepo.Update(rec); err != nil { + return + } else if err = svc.recordRepo.UpdateValues(rec.ID, rec.Values); err != nil { + return + } + case "delete": + if err = svc.recordRepo.Delete(rec); err != nil { + return err + } else if err = svc.recordRepo.DeleteValues(rec); err != nil { + return err + } + } + } + + return + }) +} + // loadCombo Loads everything we need for record manipulation // // Loads namespace, module, record and set of triggers. diff --git a/compose/service/service.go b/compose/service/service.go index 1d6c5efa7..7d140a7bc 100644 --- a/compose/service/service.go +++ b/compose/service/service.go @@ -2,6 +2,8 @@ package service import ( "context" + "errors" + "github.com/cortezaproject/corteza-server/pkg/corredor" "time" "go.uber.org/zap" @@ -141,6 +143,8 @@ func Initialize(ctx context.Context, log *zap.Logger, c Config) (err error) { DefaultNotification = Notification() DefaultAttachment = Attachment(DefaultStore) + RegisterIteratorProviders() + return nil } @@ -159,6 +163,39 @@ func Watchers(ctx context.Context) { DefaultPermissions.Watch(ctx) } +func RegisterIteratorProviders() { + // Register resource finders on iterator + corredor.Service().RegisterIteratorProvider( + "compose:record", + func(ctx context.Context, f map[string]string, h eventbus.HandlerFn, action string) error { + rf := types.RecordFilter{ + Filter: f["filter"], + Sort: f["sort"], + } + + rf.ParsePagination(f) + + if nsLookup, has := f["namespace"]; !has { + return errors.New("namespace for record iteration filter not defined") + } else if ns, err := DefaultNamespace.With(ctx).FindByAny(nsLookup); err != nil { + return err + } else { + rf.NamespaceID = ns.ID + } + + if mLookup, has := f["module"]; !has { + return errors.New("module for record iteration filter not defined") + } else if m, err := DefaultModule.With(ctx).FindByAny(rf.NamespaceID, mLookup); err != nil { + return err + } else { + rf.ModuleID = m.ID + } + + return DefaultRecord.With(ctx).Iterator(rf, h, action) + }, + ) +} + // Data is stale when new date does not match updatedAt or createdAt (before first update) func isStale(new *time.Time, updatedAt *time.Time, createdAt time.Time) bool { if new == nil {