diff --git a/compose/service/record.go b/compose/service/record.go index 9552ead8d..b155de4f9 100644 --- a/compose/service/record.go +++ b/compose/service/record.go @@ -19,6 +19,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/errors" "github.com/cortezaproject/corteza-server/pkg/eventbus" "github.com/cortezaproject/corteza-server/pkg/label" + "github.com/cortezaproject/corteza-server/pkg/report" "github.com/cortezaproject/corteza-server/store" ) @@ -85,6 +86,8 @@ type ( RecordExport(context.Context, types.RecordFilter) error RecordImport(context.Context, error) error + Datasource(context.Context, *report.LoadStepDefinition) (report.Datasource, error) + Create(ctx context.Context, record *types.Record) (*types.Record, error) Update(ctx context.Context, record *types.Record) (*types.Record, error) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (types.RecordSet, error) @@ -1373,7 +1376,6 @@ func (svc record) Iterator(ctx context.Context, f types.RecordFilter, fn eventbu }() return svc.recordAction(ctx, aProps, RecordActionIteratorInvoked, err) - } func ComposeRecordFilterChecker(ctx context.Context, ac recordAccessController, m *types.Module) func(*types.Record) (bool, error) { diff --git a/compose/service/record_datasource.go b/compose/service/record_datasource.go new file mode 100644 index 000000000..4997d003b --- /dev/null +++ b/compose/service/record_datasource.go @@ -0,0 +1,123 @@ +package service + +import ( + "context" + "fmt" + + "github.com/cortezaproject/corteza-server/compose/types" + "github.com/cortezaproject/corteza-server/pkg/errors" + "github.com/cortezaproject/corteza-server/pkg/report" + "github.com/cortezaproject/corteza-server/store" + "github.com/spf13/cast" +) + +func (svc record) Datasource(ctx context.Context, ld *report.LoadStepDefinition) (report.Datasource, error) { + var ( + moduleID uint64 + namespaceID uint64 + err error + + def = ld.Definition + ) + + if mr, ok := def["namespaceID"]; ok { + namespaceID, err = cast.ToUint64E(mr) + if err != nil { + return nil, err + } + } else if mr, ok = def["namespace"]; ok { + // slug; fetch from store + ns, err := store.LookupComposeNamespaceBySlug(ctx, svc.store, mr.(string)) + if errors.IsNotFound(err) { + err = NamespaceErrNotFound() + } + if err != nil { + return nil, err + } + namespaceID = ns.ID + } else { + return nil, fmt.Errorf("compose namespace not defined") + } + + if mr, ok := def["moduleID"]; ok { + moduleID, err = cast.ToUint64E(mr) + if err != nil { + return nil, err + } + } else if mr, ok = def["module"]; ok { + // handle; fetch from store + mod, err := store.LookupComposeModuleByNamespaceIDHandle(ctx, svc.store, namespaceID, mr.(string)) + if errors.IsNotFound(err) { + err = ModuleErrNotFound() + } + if err != nil { + return nil, err + } + moduleID = mod.ID + } + + // Find mod + mod, err := svc.store.LookupComposeModuleByID(ctx, moduleID) + if err != nil { + return nil, err + } + mod.Fields, _, err = svc.store.SearchComposeModuleFields(ctx, types.ModuleFieldFilter{ + ModuleID: []uint64{mod.ID}, + }) + if err != nil { + return nil, err + } + + if len(ld.Columns) == 0 { + cols := make(report.FrameColumnSet, 0, len(mod.Fields)+8) + + var c *report.FrameColumn + c = report.MakeColumnOfKind("Record") + c.Name = "id" + c.Label = "Record ID" + cols = append(cols, c) + + for _, f := range mod.Fields { + k := f.Kind + c = report.MakeColumnOfKind(k) + c.Name = f.Name + c.Label = f.Label + cols = append(cols, c) + } + + // Sys fields + c = report.MakeColumnOfKind("Date") + c.Name = "createdAt" + c.Label = "Created at" + cols = append(cols, c) + + c = report.MakeColumnOfKind("User") + c.Name = "createdBy" + c.Label = "Created by" + cols = append(cols, c) + + c = report.MakeColumnOfKind("Date") + c.Name = "updatedAt" + c.Label = "Updated at" + cols = append(cols, c) + + c = report.MakeColumnOfKind("User") + c.Name = "updatedBy" + c.Label = "Updated by" + cols = append(cols, c) + + c = report.MakeColumnOfKind("Date") + c.Name = "deletedAt" + c.Label = "Deleted at" + cols = append(cols, c) + + c = report.MakeColumnOfKind("User") + c.Name = "deletedBy" + c.Label = "Deleted by" + cols = append(cols, c) + + ld.Columns = cols + } + + return svc.store.ComposeRecordDatasource(ctx, mod, ld) +} diff --git a/compose/service/service.go b/compose/service/service.go index 4812f3e6d..87eafe9c7 100644 --- a/compose/service/service.go +++ b/compose/service/service.go @@ -3,6 +3,9 @@ package service import ( "context" "fmt" + "strconv" + "time" + automationService "github.com/cortezaproject/corteza-server/automation/service" "github.com/cortezaproject/corteza-server/compose/automation" "github.com/cortezaproject/corteza-server/compose/types" @@ -18,9 +21,8 @@ import ( "github.com/cortezaproject/corteza-server/pkg/objstore/plain" "github.com/cortezaproject/corteza-server/pkg/options" "github.com/cortezaproject/corteza-server/store" + systemService "github.com/cortezaproject/corteza-server/system/service" "go.uber.org/zap" - "strconv" - "time" ) type ( @@ -172,6 +174,10 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config) DefaultNamespace, ) + // Register reporters + // @todo additional datasource providers; generate? + systemService.DefaultReport.RegisterReporter("composeRecords", DefaultRecord) + return nil } diff --git a/store/compose_records.gen.go b/store/compose_records.gen.go index 32e32438e..b312f4e5b 100644 --- a/store/compose_records.gen.go +++ b/store/compose_records.gen.go @@ -10,7 +10,9 @@ package store import ( "context" + "github.com/cortezaproject/corteza-server/compose/types" + "github.com/cortezaproject/corteza-server/pkg/report" ) type ( @@ -36,6 +38,9 @@ type ( // PartialComposeRecordValueUpdate (custom function) PartialComposeRecordValueUpdate(ctx context.Context, _mod *types.Module, _values ...*types.RecordValue) error + + // @todo !!! + ComposeRecordDatasource(ctx context.Context, _mod *types.Module, _ld *report.LoadStepDefinition) (report.Datasource, error) } ) @@ -90,3 +95,7 @@ func ComposeRecordReport(ctx context.Context, s ComposeRecords, _mod *types.Modu func PartialComposeRecordValueUpdate(ctx context.Context, s ComposeRecords, _mod *types.Module, _values ...*types.RecordValue) error { return s.PartialComposeRecordValueUpdate(ctx, _mod, _values...) } + +func ComposeRecordDatasource(ctx context.Context, s ComposeRecords, _mod *types.Module, _ld *report.LoadStepDefinition) (report.Datasource, error) { + return s.ComposeRecordDatasource(ctx, _mod, _ld) +} diff --git a/store/rdbms/compose_record_datasource.go b/store/rdbms/compose_record_datasource.go new file mode 100644 index 000000000..cdb28c0e5 --- /dev/null +++ b/store/rdbms/compose_record_datasource.go @@ -0,0 +1,439 @@ +package rdbms + +import ( + "context" + "database/sql" + "fmt" + "strings" + + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/compose/types" + "github.com/cortezaproject/corteza-server/pkg/ql" + "github.com/cortezaproject/corteza-server/pkg/report" + "github.com/cortezaproject/corteza-server/pkg/slice" + "github.com/jmoiron/sqlx" +) + +type ( + recordDatasource struct { + name string + module *types.Module + + // @todo use these + supportedAggregationFunctions map[string]bool + supportedFilterFunctions map[string]bool + + store *Store + rows *sql.Rows + + baseFilter *report.RowDefinition + + cols report.FrameColumnSet + qBuilder squirrel.SelectBuilder + nestLevel int + levelColumns map[string]bool + } +) + +func ComposeRecordDatasourceBuilder(s *Store, module *types.Module, ld *report.LoadStepDefinition) (report.Datasource, error) { + var err error + + r := &recordDatasource{ + name: ld.Name, + module: module, + store: s, + cols: ld.Columns, + levelColumns: make(map[string]bool), + + supportedAggregationFunctions: slice.ToStringBoolMap([]string{ + "COUNT", + "SUM", + "MAX", + "MIN", + "AVG", + }), + + supportedFilterFunctions: slice.ToStringBoolMap([]string{ + "CONCAT", + "QUARTER", + "YEAR", + "DATE", + "NOW", + "DATE_ADD", + "DATE_SUB", + "DATE_FORMAT", + }), + } + + r.qBuilder, err = r.baseQuery(ld.Rows) + + return r, err +} + +func (r *recordDatasource) Name() string { + if r.name != "" { + return r.name + } + + if r.module == nil { + return "" + } + + if r.module.Handle == "" { + return r.module.Name + } + return r.module.Handle +} + +// @todo add Transform +// @todo try to make Group and Transform use the base query + +func (r *recordDatasource) Group(d report.GroupDefinition, name string) (bool, error) { + defer func() { + r.nestLevel++ + r.name = name + }() + + cls := r.levelColumns + r.levelColumns = make(map[string]bool) + + gCols := make(report.FrameColumnSet, 0, 10) + + q := squirrel.Select() + + // @todo allow some transformation functions within the agg. functions + parser := ql.NewParser() + parser.OnFunction = r.stdAggregationHandler + parser.OnIdent = func(i ql.Ident) (ql.Ident, error) { + if !cls[i.Value] { + return i, fmt.Errorf("column %s does not exist on level %d", i.Value, r.nestLevel) + } + + i.Value = fmt.Sprintf("l%d.%s", r.nestLevel, i.Value) + return i, nil + } + + for _, g := range d.Groups { + e, err := parser.ParseExpression(g.Expr) + if err != nil { + return false, err + } + + // @todo imply based on context + c := report.MakeColumnOfKind(g.Kind) + c.Name = g.Name + gCols = append(gCols, c) + + r.levelColumns[g.Name] = true + q = q.Column(fmt.Sprintf("(%s) as `%s`", e.String(), g.Name)). + GroupBy(e.String()) + } + + for _, c := range d.Columns { + for alias, op := range c { + o := op.GetOp() + e, err := parser.ParseExpression(fmt.Sprintf("%s(%s)", o, op[o])) + if err != nil { + return false, err + } + + // @todo imply based on context + c := report.MakeColumnOfKind("Number") + c.Name = alias + gCols = append(gCols, c) + + r.levelColumns[alias] = true + q = q. + Column(fmt.Sprintf("%s as `%s`", e.String(), alias)) + } + } + + if d.Rows != nil { + // @todo validate groupping functions + hh, err := r.rowFilterToString("", gCols, d.Rows) + if err != nil { + return false, err + } + q = q.Having(hh) + } + + r.cols = gCols + r.qBuilder = q.FromSelect(r.qBuilder, fmt.Sprintf("l%d", r.nestLevel)) + return true, nil +} + +func (r *recordDatasource) Load(ctx context.Context, dd ...*report.FrameDefinition) (l report.Loader, c report.Closer, err error) { + def := dd[0] + + // assure columns + // - undefined columns = all columns + if len(def.Columns) == 0 { + def.Columns = r.cols + } else { + // - make sure they exist + cc := make(report.FrameColumnSet, len(def.Columns)) + for i, c := range def.Columns { + ci := r.cols.Find(c.Name) + if ci == -1 { + return nil, nil, fmt.Errorf("column not found: %s", c.Name) + } + cc[i] = r.cols[ci] + } + def.Columns = cc + } + + q := r.qBuilder + + // when filtering/sorting, wrap the base query in a sub-select, so we don't need to + // worry about exact column names. + // + // @todo flatten the query + if def.Rows != nil || def.Sorting != nil { + wrap := squirrel.Select("*").FromSelect(q, "w_base") + + // additional filtering + if def.Rows != nil { + f, err := r.rowFilterToString("", r.cols, def.Rows) + if err != nil { + return nil, nil, err + } + wrap = wrap.Where(f) + } + + // additional sorting + if len(def.Sorting) > 0 { + ss := make([]string, len(def.Sorting)) + for i, c := range def.Sorting { + ci := r.cols.Find(c.Column) + if ci == -1 { + return nil, nil, fmt.Errorf("sort column not resolved: %s", c.Column) + } + + _, _, typeCast, err := r.store.config.CastModuleFieldToColumnType(r.cols[ci], c.Column) + if err != nil { + return nil, nil, err + } + + ss[i] = r.store.config.SqlSortHandler(fmt.Sprintf(typeCast, c.Column), c.Descending) + } + + wrap = wrap.OrderBy(ss...) + } + + q = wrap + } + + r.rows, err = r.store.Query(ctx, q) + if err != nil { + return nil, nil, fmt.Errorf("cannot execute query: %w", err) + } + + return func(cap int) ([]*report.Frame, error) { + f := &report.Frame{ + Name: def.Name, + Source: def.Source, + Ref: def.Ref, + } + + // fetch & convert the data + i := 0 + // @todo make it in place + f.Columns = def.Columns + f.Rows = make(report.FrameRowSet, 0, cap) + + for r.rows.Next() { + i++ + + err = r.Cast(r.rows, f) + if err != nil { + return nil, err + } + + if i >= cap { + out := []*report.Frame{f} + f = &report.Frame{} + i = 0 + return out, nil + } + } + + if i > 0 { + return []*report.Frame{f}, nil + } + return nil, nil + }, func() { + if r.rows == nil { + return + } + r.rows.Close() + }, nil +} + +// @todo handle those rv_ prefixes; for now omitted +func (r *recordDatasource) baseQuery(f *report.RowDefinition) (sqb squirrel.SelectBuilder, err error) { + var ( + joinTpl = "compose_record_value AS %s ON (%s.record_id = crd.id AND %s.name = '%s' AND %s.deleted_at IS NULL)" + + report = r.store.composeRecordsSelectBuilder(). + Where("crd.deleted_at IS NULL"). + Where("crd.module_id = ?", r.module.ID) + ) + + // Prepare all of the mod columns + // @todo make this as small as possible! + for _, f := range r.module.Fields { + report = report.LeftJoin(strings.ReplaceAll(joinTpl, "%s", f.Name)). + Column(f.Name + ".value as " + f.Name) + + r.levelColumns[f.Name] = true + } + + if f != nil { + // @todo functions and function validation + parser := ql.NewParser() + parser.OnIdent = func(i ql.Ident) (ql.Ident, error) { + if !r.levelColumns[i.Value] { + return i, fmt.Errorf("column %s does not exist on level %d", i.Value, r.nestLevel) + } + + return i, nil + } + + fl, err := r.rowFilterToString("", r.cols, f) + if err != nil { + return sqb, err + } + astq, err := parser.ParseExpression(fl) + if err != nil { + return sqb, err + } + report = report.Where(astq.String()) + } + + return report, nil +} + +func (b *recordDatasource) Cast(row sqlx.ColScanner, out *report.Frame) error { + var err error + aux := make(map[string]interface{}) + if err = sqlx.MapScan(row, aux); err != nil { + return err + } + + r := make(report.FrameRow, len(out.Columns)) + + k := "" + for i, c := range out.Columns { + k = "" + c.Name + v, ok := aux[k] + if !ok { + continue + } + + // @todo improve json value casting; I couldn't figure it out at the time + + // @todo this doesn't work... + // var aux interface{} + // err = json.Unmarshal(v.([]byte), &aux) + // if err != nil { + // return err + // } + + switch cv := v.(type) { + case []byte: + c, err := c.Caster(string(cv)) + if err != nil { + return err + } + r[i] = c + + case uint64, int64: + r[i], err = c.Caster(cv) + if err != nil { + return err + } + } + + } + + out.Rows = append(out.Rows, r) + return nil +} + +// Identifiers should be names of the fields (physical table columns OR json fields, defined in module) +func (b *recordDatasource) stdAggregationHandler(f ql.Function) (ql.ASTNode, error) { + if !b.supportedAggregationFunctions[strings.ToUpper(f.Name)] { + return f, fmt.Errorf("unsupported aggregate function %q", f.Name) + } + + return b.store.SqlFunctionHandler(f) +} + +func (ds *recordDatasource) rowFilterToString(conjunction string, cc report.FrameColumnSet, def ...*report.RowDefinition) (string, error) { + // The fields on the root level of the definition take priority + base := "" + for _, f := range def { + if f == nil { + continue + } + + if f.Cells != nil { + for k, op := range f.Cells { + ci := cc.Find(k) + if ci == -1 { + return "", fmt.Errorf("filtered column not found in the data frame: %s", k) + } + _, _, typeCast, err := ds.store.config.CastModuleFieldToColumnType(cc[ci], k) + if err != nil { + return "", err + } + + col := fmt.Sprintf(typeCast, k) + base += fmt.Sprintf("%s%s%s %s ", col, op.OpToCmp(), fmt.Sprintf(typeCast, op.Value), conjunction) + } + } + } + if strings.TrimSpace(base) != "" { + base = strings.TrimSuffix(base, " "+conjunction+" ") + } + + // Nested AND + for _, f := range def { + if f == nil { + continue + } + + if f.And != nil { + na, err := ds.rowFilterToString("AND", cc, f.And...) + if err != nil { + return "", err + } + if strings.TrimSpace(base) != "" { + base += fmt.Sprintf(" %s (%s)", conjunction, na) + } else { + base = na + } + } + } + + // Nested OR + for _, f := range def { + if f == nil { + continue + } + + if f.Or != nil { + na, err := ds.rowFilterToString("OR", cc, f.Or...) + if err != nil { + return "", err + } + if strings.TrimSpace(base) != "" { + base += fmt.Sprintf(" %s (%s)", conjunction, na) + } else { + base = na + } + } + } + + return strings.TrimSpace(base), nil +} diff --git a/store/rdbms/compose_records.go b/store/rdbms/compose_records.go index 3595bd2f1..a0fd53e7a 100644 --- a/store/rdbms/compose_records.go +++ b/store/rdbms/compose_records.go @@ -10,6 +10,7 @@ import ( "github.com/cortezaproject/corteza-server/compose/types" "github.com/cortezaproject/corteza-server/pkg/filter" "github.com/cortezaproject/corteza-server/pkg/ql" + "github.com/cortezaproject/corteza-server/pkg/report" "github.com/cortezaproject/corteza-server/store" "github.com/cortezaproject/corteza-server/store/rdbms/builders" ) @@ -447,6 +448,10 @@ func (s Store) ComposeRecordReport(ctx context.Context, m *types.Module, metrics return ComposeRecordReportBuilder(&s, m, metrics, dimensions, filter).Run(ctx) } +func (s Store) ComposeRecordDatasource(ctx context.Context, m *types.Module, ld *report.LoadStepDefinition) (report.Datasource, error) { + return ComposeRecordDatasourceBuilder(&s, m, ld) +} + func (s Store) convertComposeRecordFilter(m *types.Module, f types.RecordFilter) (query squirrel.SelectBuilder, err error) { if m == nil { err = fmt.Errorf("module not provided")