feat(dq): add create_dq_job + prepare_create_dq_job MCP tools

cmd/chip/experimental.go

@@ -17,8 +17,9 @@ import (
 // the help text, example YAML, and validation warning all read from this
 // map so no other code needs to change.
 var knownExperimentalFeatures = map[string]string{
-	skills.FeatureName:                    "Embedded skill catalog served via list_collibra_skills and load_collibra_skill.",
+	skills.FeatureName:                 "Embedded skill catalog served via list_collibra_skills and load_collibra_skill.",
 	tools.ContextSpecificationsFeature: "Context specification tools: list_context_specifications, get_context_specification, and contextSpecificationId parameter on get_asset_details.",
+	tools.DataQualityFeatureName:       "Data Quality job tools (prepare_create_data_quality_job, create_data_quality_job) that create and queue DQ jobs.",
 }
 
 // validateExperimental warns (without exiting) when the user enabled an

pkg/clients/dgc_client.go

@@ -70,6 +70,23 @@ func GetAssetSummary(
 	return ParseAssetDetailsGraphQLResponse(body)
 }
 
+// GetAssetWithRelations fetches a single asset (with its first page of incoming/outgoing relations)
+// by UUID string. Thin wrapper over GetAssetSummary for callers that have a string id.
+func GetAssetWithRelations(ctx context.Context, collibraHttpClient *http.Client, assetID string) (*Asset, error) {
+	id, err := uuid.Parse(assetID)
+	if err != nil {
+		return nil, fmt.Errorf("invalid asset id %q: %w", assetID, err)
+	}
+	assets, err := GetAssetSummary(ctx, collibraHttpClient, id, "", "")
+	if err != nil {
+		return nil, err
+	}
+	if len(assets) == 0 {
+		return nil, fmt.Errorf("asset %s not found", assetID)
+	}
+	return &assets[0], nil
+}
+
 func ListAssetTypes(ctx context.Context, collibraHttpClient *http.Client, limit int, offset int) (*AssetTypePagedResponse, error) {
 	slog.InfoContext(ctx, fmt.Sprintf("Listing asset types with limit: %d, offset: %d", limit, offset))
 

pkg/clients/dgc_dq_client_test.go

@@ -0,0 +1,129 @@
+package clients
+
+import "testing"
+
+func TestBuildProfileMonitorsMapsKeysToFields(t *testing.T) {
+	pm, unknown := BuildProfileMonitors([]string{"rowCount", "MIN", " uniqueness ", "descriptiveStatistics"})
+	if len(unknown) != 0 {
+		t.Fatalf("expected no unknown keys, got %v", unknown)
+	}
+	// Case-insensitive + trimmed keys map to the right fields.
+	if !pm.RowCount || !pm.Min || !pm.Uniqueness || !pm.DescriptiveStatistics {
+		t.Errorf("selected monitors not all ON: %+v", pm)
+	}
+	// Everything not selected is OFF.
+	if pm.NullValues || pm.EmptyFields || pm.Mean || pm.Max || pm.ExecutionTime {
+		t.Errorf("unselected monitors should be OFF: %+v", pm)
+	}
+}
+
+func TestBuildProfileMonitorsReportsUnknown(t *testing.T) {
+	_, unknown := BuildProfileMonitors([]string{"rowCount", "bogus"})
+	if len(unknown) != 1 || unknown[0] != "bogus" {
+		t.Fatalf("expected unknown=[bogus], got %v", unknown)
+	}
+}
+
+func TestDefaultMonitorKeysMatchCatalog(t *testing.T) {
+	defaults := map[string]bool{}
+	for _, k := range DefaultMonitorKeys() {
+		defaults[k] = true
+	}
+	for _, m := range DqMonitorCatalog() {
+		if m.DefaultEnabled != defaults[m.Key] {
+			t.Errorf("monitor %q: DefaultEnabled=%v but DefaultMonitorKeys membership=%v", m.Key, m.DefaultEnabled, defaults[m.Key])
+		}
+	}
+}
+
+func TestBuildSchedulingSettingsNeverIsNil(t *testing.T) {
+	for _, repeat := range []string{"", "NEVER", "never"} {
+		s, err := BuildSchedulingSettings(DqScheduleInput{Repeat: repeat})
+		if err != nil || s != nil {
+			t.Errorf("repeat=%q: expected (nil,nil), got (%+v,%v)", repeat, s, err)
+		}
+	}
+}
+
+func TestBuildSchedulingSettingsDailyDefaults(t *testing.T) {
+	s, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "DAILY"})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if s.SchedulerMode != "DAILY" || s.Daily == nil || len(s.Daily.DaysOfWeek) != 7 {
+		t.Errorf("expected DAILY with 7 days, got %+v", s)
+	}
+	if s.ScheduledRunTime != "00:00:00" || s.Daily.DailyOffset != "SCHEDULED" || !s.IsActive {
+		t.Errorf("unexpected defaults: %+v / daily %+v", s, s.Daily)
+	}
+}
+
+func TestBuildSchedulingSettingsWeekdays(t *testing.T) {
+	s, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "WEEKDAYS", RunTime: "08:00:00"})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if s.SchedulerMode != "DAILY" || len(s.Daily.DaysOfWeek) != 5 {
+		t.Errorf("expected Mon-Fri (5 days), got %+v", s.Daily)
+	}
+}
+
+func TestBuildSchedulingSettingsWeeklyInvalidDay(t *testing.T) {
+	if _, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "WEEKLY", DaysOfWeek: []string{"FUNDAY"}}); err == nil {
+		t.Errorf("expected error for invalid day FUNDAY")
+	}
+	if _, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "WEEKLY"}); err == nil {
+		t.Errorf("expected error for WEEKLY with no days")
+	}
+}
+
+func TestBuildSchedulingSettingsHourly(t *testing.T) {
+	s, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "HOURLY", RunDateOffset: "TWO_HOURS"})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if s.SchedulerMode != "HOURLY" || s.Hourly == nil || s.Hourly.HourlyOffset != "TWO_HOURS" {
+		t.Errorf("expected HOURLY/TWO_HOURS, got %+v", s)
+	}
+	if _, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "HOURLY", RunDateOffset: "SEVEN_DAYS"}); err == nil {
+		t.Errorf("expected error: SEVEN_DAYS is not a valid hourly offset")
+	}
+}
+
+func TestHasPermission(t *testing.T) {
+	perms := []string{"DATA_QUALITY_JOB_CREATE", "DATA_QUALITY_JOB_RUN"}
+	if !HasPermission(perms, "DATA_QUALITY_JOB_CREATE") {
+		t.Error("expected create permission present")
+	}
+	if !HasPermission(perms, "data_quality_job_run") { // case-insensitive
+		t.Error("expected run permission present (case-insensitive)")
+	}
+	if HasPermission(perms, "DATA_QUALITY_JOB_SCHEDULE") {
+		t.Error("did not expect schedule permission")
+	}
+}
+
+func TestDqJobDetailsPath(t *testing.T) {
+	if got := DqJobDetailsPath("sales.orders"); got != "/data-quality/jobs?jobName=sales.orders" {
+		t.Errorf("unexpected path: %q", got)
+	}
+	if p := DqJobDetailsPath("my job"); p != "/data-quality/jobs?jobName=my+job" {
+		t.Errorf("expected query-escaped name, got %q", p)
+	}
+}
+
+func TestBuildSchedulingSettingsMonthly(t *testing.T) {
+	s, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "MONTHLY", MonthlyMode: "LAST", RunDateOffset: "FIRST_OF_PRIOR_MONTH"})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if s.Monthly == nil || s.Monthly.MonthlyRepeat != "LAST" || s.Monthly.MonthlyOffset != "FIRST_OF_PRIOR_MONTH" {
+		t.Errorf("expected MONTHLY LAST/FIRST_OF_PRIOR_MONTH, got %+v", s.Monthly)
+	}
+	if _, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "MONTHLY", MonthlyMode: "DAY", DayOfMonth: 0}); err == nil {
+		t.Errorf("expected error: DAY mode needs dayOfMonth 1-28")
+	}
+	if _, err := BuildSchedulingSettings(DqScheduleInput{Repeat: "MONTHLY", MonthlyMode: "DAY", DayOfMonth: 31}); err == nil {
+		t.Errorf("expected error: dayOfMonth 31 is out of range")
+	}
+}

pkg/clients/dq_notifications.go

@@ -0,0 +1,199 @@
+package clients
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+)
+
+// DQ job notifications. Unlike rowFilter/sampleSetting (which are persisted-but-not-applied on the
+// scan), the structured `notifications` field is APPLIED: the server's JobNotificationsMapper turns
+// each NotificationOption into a real AlertCond (verified). So this is a clean structured feature —
+// no SQL/dialect involvement. Contract: JobNotifications/NotificationOption in ui-v1-private-oas-spec.yaml.
+
+// DqNotificationOption is one notification rule (→ one AlertCond server-side).
+type DqNotificationOption struct {
+	NotificationType string `json:"notificationType"`
+	Enabled          bool   `json:"enabled"`
+	Message          string `json:"message,omitempty"`
+	Quantity         int    `json:"quantity,omitempty"`
+}
+
+// DqJobNotifications is the public `notifications` object (dq-v1-public-oas-spec.yaml JobNotifications).
+// Recipients are delivered via channels (currently EMAIL) carrying platform USERNAMES — not UUIDs. The
+// public schema requires both notificationOptions and channels (>=1) when notifications are configured.
+type DqJobNotifications struct {
+	NotificationOptions   []DqNotificationOption  `json:"notificationOptions"`
+	GlobalMessage         string                  `json:"globalMessage,omitempty"`
+	UseIndividualMessages bool                    `json:"useIndividualMessages"`
+	Channels              []DqNotificationChannel `json:"channels"`
+}
+
+// DqNotificationChannel is one delivery channel and its recipients. Channel is EMAIL today; recipients
+// are platform usernames.
+type DqNotificationChannel struct {
+	Channel    string   `json:"channel"`
+	Recipients []string `json:"recipients"`
+}
+
+// DqNotificationInfo describes one selectable notification for display/selection.
+type DqNotificationInfo struct {
+	Key              string `json:"key"`
+	Label            string `json:"label"`
+	NotificationType string `json:"notificationType"`
+	DefaultEnabled   bool   `json:"defaultEnabled"`
+	TakesQuantity    bool   `json:"takesQuantity"`
+	DefaultQuantity  int    `json:"defaultQuantity,omitempty"`
+}
+
+// DqNotificationCatalog is the selectable notifications with wizard defaults (Step Notifications):
+// Job failed, Rows<=, Score<=, Run time> ON; Job completed, Runs/Days without data OFF.
+func DqNotificationCatalog() []DqNotificationInfo {
+	return []DqNotificationInfo{
+		{Key: "jobFailed", Label: "Job failed", NotificationType: "JOB_FAILED", DefaultEnabled: true},
+		{Key: "rowsBelow", Label: "Rows below limit", NotificationType: "ROWS_LESS_THAN_LIMIT", DefaultEnabled: true, TakesQuantity: true, DefaultQuantity: 1},
+		{Key: "scoreBelow", Label: "Score below limit", NotificationType: "SCORE_LESS_THAN_LIMIT", DefaultEnabled: true, TakesQuantity: true, DefaultQuantity: 75},
+		{Key: "runTimeAbove", Label: "Run time above (minutes)", NotificationType: "RUN_TIME_MORE_THEN_LIMIT", DefaultEnabled: true, TakesQuantity: true, DefaultQuantity: 60},
+		{Key: "jobCompleted", Label: "Job completed", NotificationType: "JOB_COMPLETED", DefaultEnabled: false},
+		{Key: "runsWithoutData", Label: "Runs without data", NotificationType: "RUNS_WITHOUT_DATA", DefaultEnabled: false, TakesQuantity: true, DefaultQuantity: 1},
+		{Key: "daysWithoutData", Label: "Days without data", NotificationType: "DAYS_WITHOUT_DATA", DefaultEnabled: false, TakesQuantity: true, DefaultQuantity: 1},
+	}
+}
+
+// NotificationKeys returns every valid notification key, in catalog order.
+func NotificationKeys() []string {
+	cat := DqNotificationCatalog()
+	keys := make([]string, 0, len(cat))
+	for _, n := range cat {
+		keys = append(keys, n.Key)
+	}
+	return keys
+}
+
+// DefaultNotificationKeys returns the keys enabled by default.
+func DefaultNotificationKeys() []string {
+	var keys []string
+	for _, n := range DqNotificationCatalog() {
+		if n.DefaultEnabled {
+			keys = append(keys, n.Key)
+		}
+	}
+	return keys
+}
+
+// BuildNotificationOptions turns enabled keys (case-insensitive) into NotificationOptions. quantities
+// overrides a key's threshold (keyed by lower-case key; <=0 uses the catalog default). messages sets a
+// per-notification message (keyed by lower-case key) that overrides the global message for that one
+// notification — pass nil for none. Unknown keys are returned so the caller can reject them.
+func BuildNotificationOptions(enabledKeys []string, quantities map[string]int, messages map[string]string) ([]DqNotificationOption, []string) {
+	valid := map[string]DqNotificationInfo{}
+	for _, n := range DqNotificationCatalog() {
+		valid[strings.ToLower(n.Key)] = n
+	}
+	var opts []DqNotificationOption
+	var unknown []string
+	for _, k := range enabledKeys {
+		lk := strings.ToLower(strings.TrimSpace(k))
+		if lk == "" {
+			continue
+		}
+		info, ok := valid[lk]
+		if !ok {
+			unknown = append(unknown, k)
+			continue
+		}
+		opt := DqNotificationOption{NotificationType: info.NotificationType, Enabled: true}
+		if info.TakesQuantity {
+			q := quantities[lk]
+			if q <= 0 {
+				q = info.DefaultQuantity
+			}
+			opt.Quantity = q
+		}
+		if m := strings.TrimSpace(messages[lk]); m != "" {
+			opt.Message = m
+		}
+		opts = append(opts, opt)
+	}
+	return opts, unknown
+}
+
+// RecipientResolution is the outcome of resolving notification recipients to active users. UserIDs and
+// Usernames are positionally aligned (same resolved user); Usernames feed the public notification
+// channels (which take usernames), UserIDs are kept for callers that need the UUID.
+type RecipientResolution struct {
+	UserIDs    []string
+	Usernames  []string
+	Unresolved []string
+}
+
+// ResolveNotificationRecipients resolves each username/email to an active user's UUID. An entry
+// containing '@' is looked up by email, otherwise by username. Not-found or disabled accounts land
+// in Unresolved (the list endpoint excludes disabled users and the email lookup 404s), so the
+// caller can warn and decide whether to proceed without them. Duplicates are de-duped.
+func ResolveNotificationRecipients(ctx context.Context, client *http.Client, recipients []string) (RecipientResolution, error) {
+	var res RecipientResolution
+	seen := map[string]bool{}
+	for _, r := range recipients {
+		r = strings.TrimSpace(r)
+		if r == "" {
+			continue
+		}
+		var (
+			u   *EditAssetUser
+			err error
+		)
+		if strings.Contains(r, "@") {
+			u, err = FindUserByEmail(ctx, client, r)
+		} else {
+			u, err = FindUserByUsername(ctx, client, r)
+		}
+		if err != nil {
+			return res, err
+		}
+		if u == nil || u.ID == "" {
+			res.Unresolved = append(res.Unresolved, r)
+			continue
+		}
+		if !seen[u.ID] {
+			seen[u.ID] = true
+			res.UserIDs = append(res.UserIDs, u.ID)
+			username := strings.TrimSpace(u.UserName)
+			if username == "" {
+				username = r // fall back to what the caller provided if the account has no username
+			}
+			res.Usernames = append(res.Usernames, username)
+		}
+	}
+	return res, nil
+}
+
+// GetCurrentUser returns the invoking user (GET /rest/2.0/users/current) — the default notification
+// recipient.
+func GetCurrentUser(ctx context.Context, client *http.Client) (*EditAssetUser, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "/rest/2.0/users/current", nil)
+	if err != nil {
+		return nil, fmt.Errorf("get current user: building request: %w", err)
+	}
+	req.Header.Set("Accept", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("get current user: sending request: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("get current user: reading response: %w", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("get current user: status %d: %s", resp.StatusCode, string(body))
+	}
+	var user EditAssetUser
+	if err := json.Unmarshal(body, &user); err != nil {
+		return nil, fmt.Errorf("get current user: decoding response: %w", err)
+	}
+	return &user, nil
+}