package service import ( "context" "fmt" "strconv" "strings" "github.com/cortezaproject/corteza/server/pkg/dal" "github.com/cortezaproject/corteza/server/pkg/errors" "github.com/cortezaproject/corteza/server/pkg/locale" "github.com/cortezaproject/corteza/server/pkg/actionlog" "github.com/cortezaproject/corteza/server/pkg/filter" "github.com/cortezaproject/corteza/server/pkg/label" "github.com/cortezaproject/corteza/server/store" "github.com/cortezaproject/corteza/server/system/reporting" "github.com/cortezaproject/corteza/server/system/types" "github.com/modern-go/reflect2" "github.com/spf13/cast" ) type ( report struct { ac reportAccessController eventbus eventDispatcher actionlog actionlog.Recorder store store.Storer locale locale.Locale users UserService pipelineRunner pipelineRunner } reportAccessController interface { CanSearchReports(context.Context) bool CanCreateReport(context.Context) bool CanReadReport(context.Context, *types.Report) bool CanUpdateReport(context.Context, *types.Report) bool CanDeleteReport(context.Context, *types.Report) bool CanRunReport(context.Context, *types.Report) bool } pipelineRunner interface { Run(context.Context, dal.Pipeline) (dal.Iterator, error) Dryrun(context.Context, dal.Pipeline) error FindModel(dal.ModelRef) *dal.Model } ) // Report is a default report service initializer func Report(s store.Storer, ac reportAccessController, al actionlog.Recorder, eb eventDispatcher) *report { return &report{ store: s, ac: ac, actionlog: al, eventbus: eb, locale: locale.Global(), users: DefaultUser, pipelineRunner: dal.Service(), } } func (svc *report) LookupByID(ctx context.Context, ID uint64) (report *types.Report, err error) { var ( aaProps = &reportActionProps{report: &types.Report{ID: ID}} ) err = func() error { if report, err = loadReport(ctx, svc.store, ID); err != nil { return ReportErrInvalidID().Wrap(err) } if !svc.ac.CanReadReport(ctx, report) { return ReportErrNotAllowedToRead() } return nil }() return report, svc.recordAction(ctx, aaProps, ReportActionLookup, err) } func (svc *report) Search(ctx context.Context, rf types.ReportFilter) (rr types.ReportSet, f types.ReportFilter, err error) { var ( aaProps = &reportActionProps{filter: &rf} ) // For each fetched item, store backend will check if it is valid or not rf.Check = func(res *types.Report) (bool, error) { if !svc.ac.CanReadReport(ctx, res) { return false, nil } return true, nil } err = func() error { if !svc.ac.CanSearchReports(ctx) { return ReportErrNotAllowedToSearch() } if len(rf.Labels) > 0 { rf.LabeledIDs, err = label.Search( ctx, svc.store, types.Report{}.LabelResourceKind(), rf.Labels, ) if err != nil { return err } // labels specified but no labeled resources found if len(rf.LabeledIDs) == 0 { return nil } } if rr, f, err = store.SearchReports(ctx, svc.store, rf); err != nil { return err } if err = label.Load(ctx, svc.store, toLabeledReports(rr)...); err != nil { return err } return nil }() return rr, f, svc.recordAction(ctx, aaProps, ReportActionSearch, err) } func (svc *report) Create(ctx context.Context, new *types.Report) (report *types.Report, err error) { var ( aaProps = &reportActionProps{report: new} ) err = func() (err error) { if !svc.ac.CanCreateReport(ctx) { return ReportErrNotAllowedToCreate() } // if err = svc.eventbus.WaitFor(ctx, event.ReportBeforeCreate(new, nil)); err != nil { // return // } // Set new values after beforeCreate events are emitted new.ID = nextID() new.CreatedAt = *now() if new.Meta == nil { new.Meta = &types.ReportMeta{} } new = svc.setIDs(new) aaProps.setNew(new) if err = store.CreateReport(ctx, svc.store, new); err != nil { return } if err = label.Create(ctx, svc.store, new); err != nil { return } report = new // _ = svc.eventbus.WaitFor(ctx, event.ReportAfterCreate(new, nil)) return nil }() return report, svc.recordAction(ctx, aaProps, ReportActionCreate, err) } func (svc *report) Update(ctx context.Context, upd *types.Report) (report *types.Report, err error) { var ( aaProps = &reportActionProps{update: upd} ) err = func() (err error) { if report, err = loadReport(ctx, svc.store, upd.ID); err != nil { return } aaProps.setReport(report) if !svc.ac.CanUpdateReport(ctx, report) { return ReportErrNotAllowedToUpdate() } // Test if stale (update has an older version of data) if isStale(upd.UpdatedAt, report.UpdatedAt, report.CreatedAt) { return ReportErrStaleData() } // if err = svc.eventbus.WaitFor(ctx, event.ReportBeforeUpdate(upd, report)); err != nil { // return // } // Assign changed values after afterUpdate events are emitted report.Handle = upd.Handle report.Meta = upd.Meta report.Scenarios = upd.Scenarios report.Sources = upd.Sources report.Blocks = upd.Blocks report.UpdatedAt = now() if upd.Meta != nil { report.Meta = upd.Meta } report = svc.setIDs(report) if err = store.UpdateReport(ctx, svc.store, report); err != nil { return err } if label.Changed(report.Labels, upd.Labels) { if err = label.Update(ctx, svc.store, upd); err != nil { return } report.Labels = upd.Labels } // _ = svc.eventbus.WaitFor(ctx, event.ReportAfterUpdate(upd, report)) return nil }() return report, svc.recordAction(ctx, aaProps, ReportActionUpdate, err) } func (svc *report) Delete(ctx context.Context, ID uint64) (err error) { var ( aaProps = &reportActionProps{} report *types.Report ) err = func() (err error) { if report, err = loadReport(ctx, svc.store, ID); err != nil { return } aaProps.setReport(report) if !svc.ac.CanDeleteReport(ctx, report) { return ReportErrNotAllowedToDelete() } // if err = svc.eventbus.WaitFor(ctx, event.ReportBeforeDelete(nil, report)); err != nil { // return // } report.DeletedAt = now() if err = store.UpdateReport(ctx, svc.store, report); err != nil { return } // _ = svc.eventbus.WaitFor(ctx, event.ReportAfterDelete(nil, report)) return nil }() return svc.recordAction(ctx, aaProps, ReportActionDelete, err) } func (svc *report) Undelete(ctx context.Context, ID uint64) (err error) { var ( aaProps = &reportActionProps{} report *types.Report ) err = func() (err error) { if report, err = loadReport(ctx, svc.store, ID); err != nil { return } aaProps.setReport(report) if !svc.ac.CanDeleteReport(ctx, report) { return ReportErrNotAllowedToUndelete() } // if err = svc.eventbus.WaitFor(ctx, event.ReportBeforeUndelete(nil, app)); err != nil { // return // } report.DeletedAt = nil if err = store.UpdateReport(ctx, svc.store, report); err != nil { return } // _ = svc.eventbus.WaitFor(ctx, event.ReportAfterUndelete(nil, app)) return nil }() return svc.recordAction(ctx, aaProps, ReportActionUndelete, err) } // @todo actionlog? func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out []reporting.FrameDescription, err error) { out = make([]reporting.FrameDescription, 0, len(sources)*2) err = func() (err error) { if !svc.ac.CanCreateReport(ctx) { return ReportErrNotAllowedToCreate() } ss := src.ReportSteps() ss = append(ss, st...) out, err = reporting.Describe(ctx, svc.pipelineRunner, ss, sources) return err }() return out, err } func (svc *report) Run(ctx context.Context, reportID uint64, dd reporting.FrameDefinitionSet) (_ []*reporting.Frame, err error) { var ( aaProps = &reportActionProps{} iter dal.Iterator ff []*reporting.Frame out = make([]*reporting.Frame, 0, 4) ) err = func() (err error) { // Load the report r, err := loadReport(ctx, svc.store, reportID) if err != nil { return err } // Access control if !svc.ac.CanRunReport(ctx, r) { return ReportErrNotAllowedToRun() } // Get all of the steps ss := r.Sources.ReportSteps() ss = append(ss, r.Blocks.ReportSteps()...) // Prepare a set of runs for the provided definitions runs, err := reporting.Runs(svc.pipelineRunner, ss, dd) if err != nil { return } // Run the reports and produce the frames // @todo this can be ran in paralel for _, run := range runs { err = func() (err error) { iter, err = svc.pipelineRunner.Run(ctx, run.Pipeline) if err != nil { return } defer iter.Close() ff, err = reporting.Frames(ctx, iter, run) if err != nil { return } err = svc.enhance(ctx, ff) if err != nil { return } out = append(out, ff...) return }() if err != nil { return } } return nil }() return out, svc.recordAction(ctx, aaProps, ReportActionRun, err) } // enhance is a temporary function that enriches the output to satisfy some current requirements. // @todo extend core implementation to support such operatons // // - userID is replaced by the user name || username || email || handle || userID func (svc *report) enhance(ctx context.Context, ff []*reporting.Frame) (err error) { // Preload sys users uIndex := make(map[uint64]*types.User) uu, uf, err := svc.users.Find(ctx, types.UserFilter{Paging: filter.Paging{Limit: 1024}}) if err != nil { return } hasMore := uf.NextPage != nil for i := range uu { uIndex[uu[i].ID] = uu[i] } var uID uint64 for _, f := range ff { userCols := make([]int, 0, len(f.Columns)) for i, c := range f.Columns { // Translate system columns if c.System { pp := strings.Split(c.Name, ".") c.Label = svc.locale.T(ctx, "compose", fmt.Sprintf("field.system.%s", pp[len(pp)-1])) f.Columns[i] = c } // Collect user columns to replace IDs with labels if c.Kind != "User" { continue } userCols = append(userCols, i) } for _, r := range f.Rows { for _, ci := range userCols { col := r[ci] if reflect2.IsNil(col) { continue } uID, err = cast.ToUint64E(col) if err != nil { continue } user, ok := uIndex[uID] if !ok && hasMore { user, err = svc.users.FindByID(ctx, uID) if err != nil && err != store.ErrNotFound { return } } if user == nil { continue } else if _, ok := uIndex[uID]; !ok { uIndex[uID] = user } if usr, ok := uIndex[uID]; ok { r[ci] = strconv.FormatUint(uID, 10) if usr.Name != "" { r[ci] = usr.Name } else if usr.Username != "" { r[ci] = usr.Username } else if usr.Email != "" { r[ci] = usr.Email } else if usr.Handle != "" { r[ci] = usr.Handle } } } } } return err } func (svc *report) setIDs(r *types.Report) *types.Report { // scenarios for _, s := range r.Scenarios { if s.ScenarioID == 0 { s.ScenarioID = nextID() } } // blocks for _, b := range r.Blocks { if b.BlockID == 0 { b.BlockID = nextID() } // elements for _, elRaw := range b.Elements { el, ok := elRaw.(map[string]interface{}) if !ok { continue } elID, ok := el["elementID"] sElID := cast.ToString(elID) if sElID != "" && sElID != "0" { continue } if cast.ToUint64(elID) != 0 { continue } el["elementID"] = strconv.FormatUint(nextID(), 10) } } return r } func loadReport(ctx context.Context, s store.Reports, ID uint64) (res *types.Report, err error) { if ID == 0 { return nil, ReportErrInvalidID() } if res, err = store.LookupReportByID(ctx, s, ID); errors.IsNotFound(err) { return nil, ReportErrNotFound() } return } // toLabeledReports converts to []label.LabeledResource // // This function is auto-generated. func toLabeledReports(set []*types.Report) []label.LabeledResource { if len(set) == 0 { return nil } ll := make([]label.LabeledResource, len(set)) for i := range set { ll[i] = set[i] } return ll }