// Package datahub provides a DataHub implementation of the semantic provider.
package datahub

import (
	"context"
	"fmt"
	"strings"
	"time"

	dhclient "github.com/txn2/mcp-datahub/pkg/client"
	"github.com/txn2/mcp-datahub/pkg/types"

	"github.com/txn2/mcp-data-platform/pkg/semantic"
)

const (
	// defaultPlatform is the default data platform for URN building.
	defaultPlatform = "trino"

	// defaultTimeout is the default HTTP client timeout.
	defaultTimeout = 30 * time.Second

	// urnPartsMinCount is the minimum number of dot-separated parts in a URN name
	// (e.g., "schema.table" = 2, "catalog.schema.table" = 3).
	urnPartsMinCount = 3

	// datahubPlatform is the provider name for this adapter.
	datahubPlatform = "datahub"

	// DataHub search-filter field names. Defined as constants because
	// each appears in multiple places (filter mapping, lineage inheritance,
	// query construction).
	filterFieldPlatform = "platform"
	filterFieldTags     = "tags"
	filterFieldOwners   = "owners"
)

// Config holds DataHub adapter configuration.
type Config struct {
	URL      string
	Token    string
	Platform string // Default platform for URN building (e.g., "trino", "postgres")
	Timeout  time.Duration
	Debug    bool // Enable debug logging

	// CatalogMapping maps query engine catalog names to metadata catalog names.
	// For example: {"rdbms": "warehouse"} means the Trino "rdbms" catalog
	// corresponds to the "warehouse" catalog in DataHub URNs.
	CatalogMapping map[string]string

	// Lineage configuration for inheritance-aware column resolution.
	Lineage LineageConfig
}

// Client defines the interface for DataHub operations.
// This allows for mocking in tests.
type Client interface {
	SearchAcrossEntities(ctx context.Context, query string, opts ...dhclient.SearchOption) (*types.SearchResult, error)
	SemanticSearch(ctx context.Context, query string, opts ...dhclient.SearchOption) (*types.SearchResult, error)
	GetEntity(ctx context.Context, urn string) (*types.Entity, error)
	GetSchema(ctx context.Context, urn string) (*types.SchemaMetadata, error)
	GetSchemas(ctx context.Context, urns []string) (map[string]*types.SchemaMetadata, error)
	GetLineage(ctx context.Context, urn string, opts ...dhclient.LineageOption) (*types.LineageResult, error)
	GetColumnLineage(ctx context.Context, urn string) (*types.ColumnLineage, error)
	GetGlossaryTerm(ctx context.Context, urn string) (*types.GlossaryTerm, error)
	GetQueries(ctx context.Context, urn string) (*types.QueryList, error)
	Ping(ctx context.Context) error
	Close() error
}

// Adapter implements semantic.Provider using DataHub.
type Adapter struct {
	cfg       Config
	client    Client
	sanitizer *semantic.Sanitizer
}

// New creates a new DataHub adapter with a real client.
func New(cfg Config) (*Adapter, error) {
	if cfg.URL == "" {
		return nil, fmt.Errorf("datahub URL is required")
	}
	if cfg.Platform == "" {
		cfg.Platform = defaultPlatform
	}
	if cfg.Timeout == 0 {
		cfg.Timeout = defaultTimeout
	}

	clientCfg := dhclient.DefaultConfig()
	clientCfg.URL = cfg.URL
	clientCfg.Token = cfg.Token
	clientCfg.Timeout = cfg.Timeout
	clientCfg.Debug = cfg.Debug

	client, err := dhclient.New(clientCfg)
	if err != nil {
		return nil, fmt.Errorf("creating datahub client: %w", err)
	}

	return &Adapter{
		cfg:       cfg,
		client:    client,
		sanitizer: semantic.NewSanitizer(semantic.DefaultSanitizeConfig()),
	}, nil
}

// NewWithClient creates a new DataHub adapter with a provided client (for testing).
func NewWithClient(cfg Config, client Client) (*Adapter, error) {
	if client == nil {
		return nil, fmt.Errorf("datahub client is required")
	}
	if cfg.Platform == "" {
		cfg.Platform = defaultPlatform
	}
	return &Adapter{
		cfg:       cfg,
		client:    client,
		sanitizer: semantic.NewSanitizer(semantic.DefaultSanitizeConfig()),
	}, nil
}

// Name returns the provider name.
func (*Adapter) Name() string {
	return datahubPlatform
}

// LineageConfig returns the lineage configuration.
// This allows verifying that configuration was wired correctly.
func (a *Adapter) LineageConfig() LineageConfig {
	return a.cfg.Lineage
}

// GetTableContext retrieves table context from DataHub.
func (a *Adapter) GetTableContext(ctx context.Context, table semantic.TableIdentifier) (*semantic.TableContext, error) {
	urn := a.buildDatasetURN(table)

	entity, err := a.client.GetEntity(ctx, urn)
	if err != nil {
		return nil, fmt.Errorf("getting entity from datahub: %w", err)
	}

	tc := a.entityToTableContext(entity)
	return a.sanitizer.SanitizeTableContext(tc), nil
}

// GetColumnContext retrieves column context from DataHub.
func (a *Adapter) GetColumnContext(ctx context.Context, column semantic.ColumnIdentifier) (*semantic.ColumnContext, error) {
	urn := a.buildDatasetURN(column.TableIdentifier)

	schema, err := a.client.GetSchema(ctx, urn)
	if err != nil {
		return nil, fmt.Errorf("getting schema from datahub: %w", err)
	}

	// Find the column in the schema by FieldPath
	for _, field := range schema.Fields {
		fieldName := extractFieldName(field.FieldPath)
		if fieldName == column.Column {
			cc := a.fieldToColumnContext(field)
			return a.sanitizer.SanitizeColumnContext(cc), nil
		}
	}

	return nil, fmt.Errorf("column %s not found in schema", column.Column)
}

// GetColumnsContext retrieves all columns context from DataHub.
// When lineage is enabled, it inherits metadata from upstream datasets for undocumented columns.
func (a *Adapter) GetColumnsContext(ctx context.Context, table semantic.TableIdentifier) (map[string]*semantic.ColumnContext, error) {
	urn := a.buildDatasetURN(table)

	// Use lineage-aware resolution if enabled
	if a.cfg.Lineage.Enabled {
		resolver := newLineageResolver(a.client, a.cfg.Lineage, a.sanitizer)
		columns, err := resolver.resolveColumnsWithLineage(ctx, urn, table.String())
		if err != nil {
			return nil, fmt.Errorf("getting columns with lineage from datahub: %w", err)
		}
		return columns, nil
	}

	// Standard resolution without lineage
	schema, err := a.client.GetSchema(ctx, urn)
	if err != nil {
		return nil, fmt.Errorf("getting schema from datahub: %w", err)
	}

	columns := make(map[string]*semantic.ColumnContext, len(schema.Fields))
	for _, field := range schema.Fields {
		fieldName := extractFieldName(field.FieldPath)
		cc := a.fieldToColumnContext(field)
		columns[fieldName] = a.sanitizer.SanitizeColumnContext(cc)
	}

	return columns, nil
}

// GetLineage retrieves lineage from DataHub.
func (a *Adapter) GetLineage(ctx context.Context, table semantic.TableIdentifier, direction semantic.LineageDirection, maxDepth int) (*semantic.LineageInfo, error) {
	urn := a.buildDatasetURN(table)

	dhDirection := dhclient.LineageDirectionUpstream
	if direction == semantic.LineageDownstream {
		dhDirection = dhclient.LineageDirectionDownstream
	}

	result, err := a.client.GetLineage(ctx, urn,
		dhclient.WithDirection(dhDirection),
		dhclient.WithDepth(maxDepth),
	)
	if err != nil {
		return nil, fmt.Errorf("getting lineage from datahub: %w", err)
	}

	return a.lineageResultToInfo(result, direction, maxDepth), nil
}

// GetGlossaryTerm retrieves a glossary term from DataHub.
func (a *Adapter) GetGlossaryTerm(ctx context.Context, urn string) (*semantic.GlossaryTerm, error) {
	term, err := a.client.GetGlossaryTerm(ctx, urn)
	if err != nil {
		return nil, fmt.Errorf("getting glossary term from datahub: %w", err)
	}

	if term == nil {
		return nil, fmt.Errorf("glossary term not found: %s", urn)
	}

	return &semantic.GlossaryTerm{
		URN:         term.URN,
		Name:        a.sanitizer.SanitizeString(term.Name),
		Description: a.sanitizer.SanitizeDescription(term.Description),
	}, nil
}

// SearchTables searches for tables in DataHub using searchAcrossEntities.
// Supports advanced field-level filtering (column names, tags, etc.) via SearchFilter.Filters.
func (a *Adapter) SearchTables(ctx context.Context, filter semantic.SearchFilter) ([]semantic.TableSearchResult, error) {
	opts := buildSearchOptions(filter)

	var (
		result *types.SearchResult
		err    error
	)
	if strings.EqualFold(filter.Mode, "semantic") {
		result, err = a.client.SemanticSearch(ctx, filter.Query, opts...)
	} else {
		result, err = a.client.SearchAcrossEntities(ctx, filter.Query, opts...)
	}
	if err != nil {
		return nil, fmt.Errorf("searching datahub: %w", err)
	}

	results := make([]semantic.TableSearchResult, 0, len(result.Entities))
	for _, entity := range result.Entities {
		results = append(results, a.toSearchResult(entity))
	}

	return results, nil
}

// buildSearchOptions converts a semantic.SearchFilter to datahub client search options.
func buildSearchOptions(filter semantic.SearchFilter) []dhclient.SearchOption {
	var opts []dhclient.SearchOption
	if filter.Limit > 0 {
		opts = append(opts, dhclient.WithLimit(filter.Limit))
	}
	if filter.Offset > 0 {
		opts = append(opts, dhclient.WithOffset(filter.Offset))
	}

	// Entity types: explicit types, or default to DATASET.
	if len(filter.EntityTypes) > 0 {
		opts = append(opts, dhclient.WithTypes(filter.EntityTypes))
	} else {
		opts = append(opts, dhclient.WithEntityType(dhclient.DefaultEntityType))
	}

	// Build advanced filters from both legacy fields and new Filters slice.
	dhFilters := buildDHFilters(filter)
	if len(dhFilters) > 0 {
		opts = append(opts, dhclient.WithSearchFilters(dhFilters))
	}

	return opts
}

// buildDHFilters converts semantic filter fields to datahub client SearchFilter entries.
// Legacy fields (Platform, Tags, Domain, Owner) are mapped to their DataHub equivalents,
// then any explicit Filters are appended.
func buildDHFilters(filter semantic.SearchFilter) []dhclient.SearchFilter {
	var out []dhclient.SearchFilter

	if filter.Platform != "" {
		out = append(out, dhclient.SearchFilter{
			Field:  filterFieldPlatform,
			Values: []string{filter.Platform},
		})
	}
	if len(filter.Tags) > 0 {
		out = append(out, dhclient.SearchFilter{
			Field:  filterFieldTags,
			Values: filter.Tags,
		})
	}
	if filter.Domain != "" {
		out = append(out, dhclient.SearchFilter{
			Field:  "domains",
			Values: []string{filter.Domain},
		})
	}
	if filter.Owner != "" {
		out = append(out, dhclient.SearchFilter{
			Field:  filterFieldOwners,
			Values: []string{filter.Owner},
		})
	}

	// Append explicit advanced filters (fieldPaths, fieldTags, etc.)
	for _, f := range filter.Filters {
		out = append(out, dhclient.SearchFilter{
			Field:     f.Field,
			Values:    f.Values,
			Condition: f.Condition,
			Negated:   f.Negated,
		})
	}

	return out
}

// toSearchResult converts a datahub search entity to a semantic TableSearchResult.
func (a *Adapter) toSearchResult(entity types.SearchEntity) semantic.TableSearchResult {
	matchedField := ""
	if len(entity.MatchedFields) > 0 {
		matchedField = entity.MatchedFields[0].Name
	}

	domainName := ""
	if entity.Domain != nil {
		domainName = a.sanitizer.SanitizeString(entity.Domain.Name)
	}

	rawTags := make([]string, len(entity.Tags))
	for i, tag := range entity.Tags {
		rawTags[i] = tag.Name
	}

	return semantic.TableSearchResult{
		URN:          entity.URN,
		Name:         entity.Name,
		Platform:     entity.Platform,
		Description:  a.sanitizer.SanitizeDescription(entity.Description),
		Tags:         a.sanitizer.SanitizeTags(rawTags),
		Domain:       domainName,
		MatchedField: matchedField,
	}
}

// GetCuratedQueryCount returns the number of curated/saved queries for a dataset.
func (a *Adapter) GetCuratedQueryCount(ctx context.Context, urn string) (int, error) {
	queries, err := a.client.GetQueries(ctx, urn)
	if err != nil {
		return 0, fmt.Errorf("getting curated query count: %w", err)
	}
	return queries.Total, nil
}

// Close releases resources.
func (a *Adapter) Close() error {
	if a.client != nil {
		if err := a.client.Close(); err != nil {
			return fmt.Errorf("closing datahub client: %w", err)
		}
	}
	return nil
}

// buildDatasetURN creates a DataHub URN for a table.
// It applies catalog mapping to translate query engine catalogs to metadata catalogs.
func (a *Adapter) buildDatasetURN(table semantic.TableIdentifier) string {
	// DataHub URN format: urn:li:dataset:(urn:li:dataPlatform:platform,catalog.schema.table,PROD)

	catalog := table.Catalog

	// Apply catalog mapping if configured
	if mapped, ok := a.cfg.CatalogMapping[catalog]; ok {
		catalog = mapped
	}

	// Build name from available parts
	var parts []string
	if catalog != "" {
		parts = append(parts, catalog)
	}
	if table.Schema != "" {
		parts = append(parts, table.Schema)
	}
	parts = append(parts, table.Table)

	return fmt.Sprintf("urn:li:dataset:(urn:li:dataPlatform:%s,%s,PROD)", a.cfg.Platform, strings.Join(parts, "."))
}

// ResolveURN converts a DataHub URN to a table identifier.
func (*Adapter) ResolveURN(_ context.Context, urn string) (*semantic.TableIdentifier, error) {
	// Parse URN format: urn:li:dataset:(urn:li:dataPlatform:platform,name,env)
	if !strings.HasPrefix(urn, "urn:li:dataset:") {
		return nil, fmt.Errorf("invalid dataset URN: %s", urn)
	}

	// Extract the name part
	start := strings.Index(urn, ",")
	end := strings.LastIndex(urn, ",")
	if start == -1 || end == -1 || start == end {
		return nil, fmt.Errorf("invalid URN format: %s", urn)
	}

	name := urn[start+1 : end]
	parts := strings.Split(name, ".")

	switch len(parts) {
	case 2:
		return &semantic.TableIdentifier{
			Schema: parts[0],
			Table:  parts[1],
		}, nil
	case urnPartsMinCount:
		return &semantic.TableIdentifier{
			Catalog: parts[0],
			Schema:  parts[1],
			Table:   parts[2],
		}, nil
	default:
		return nil, fmt.Errorf("invalid table name in URN: %s", name)
	}
}

// BuildURN creates a URN from a table identifier.
func (a *Adapter) BuildURN(_ context.Context, table semantic.TableIdentifier) (string, error) {
	return a.buildDatasetURN(table), nil
}

// entityToTableContext converts a DataHub entity to semantic table context.
func (a *Adapter) entityToTableContext(entity *types.Entity) *semantic.TableContext {
	// Log any injection attempts in user-provided content
	a.logInjectionAttempts(entity)

	tc := &semantic.TableContext{
		URN:                  entity.URN,
		Description:          entity.Description,
		Owners:               convertOwners(entity.Owners),
		Tags:                 convertTags(entity.Tags),
		GlossaryTerms:        convertGlossaryTerms(entity.GlossaryTerms),
		Domain:               convertDomain(entity.Domain),
		Deprecation:          convertDeprecation(entity.Deprecation),
		CustomProperties:     convertProperties(entity.Properties),
		LastModified:         convertTimestamp(entity.LastModified),
		StructuredProperties: convertStructuredProperties(entity.StructuredProperties),
		DataContract:         convertDataContract(entity.DataContract),
	}

	// Populate incidents from enriched entity response (DataHub 1.4.x)
	tc.ActiveIncidents, tc.Incidents = convertIncidents(entity.ActiveIncidents)

	return tc
}

// logInjectionAttempts checks for and logs any prompt injection attempts in entity fields.
func (a *Adapter) logInjectionAttempts(entity *types.Entity) {
	logger := semantic.DefaultInjectionLogger

	// Check description
	logger.DetectAndLog(a.sanitizer, entity.URN, "description", entity.Description)

	// Check owner names
	for i, owner := range entity.Owners {
		logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("owners[%d].name", i), owner.Name)
	}

	// Check glossary term descriptions
	for i, term := range entity.GlossaryTerms {
		logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("glossaryTerms[%d].description", i), term.Description)
	}

	// Check domain description
	if entity.Domain != nil {
		logger.DetectAndLog(a.sanitizer, entity.URN, "domain.description", entity.Domain.Description)
	}

	// Check deprecation note
	if entity.Deprecation != nil {
		logger.DetectAndLog(a.sanitizer, entity.URN, "deprecation.note", entity.Deprecation.Note)
	}

	// Check custom properties
	for key, value := range entity.Properties {
		if str, ok := value.(string); ok {
			logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("properties[%s]", key), str)
		}
	}

	a.logInjectionAttemptsV14(entity)
}

// logInjectionAttemptsV14 checks DataHub 1.4.x fields for prompt injection attempts.
func (a *Adapter) logInjectionAttemptsV14(entity *types.Entity) {
	logger := semantic.DefaultInjectionLogger

	// Check structured property display names and string values
	for i, sp := range entity.StructuredProperties {
		if sp.Definition != nil {
			logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("structuredProperties[%d].displayName", i), sp.Definition.DisplayName)
		}
		for j, v := range sp.Values {
			if str, ok := v.(string); ok {
				logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("structuredProperties[%d].values[%d]", i, j), str)
			}
		}
	}

	// Check incident titles and descriptions
	if entity.ActiveIncidents != nil {
		for i, inc := range entity.ActiveIncidents.Incidents {
			logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("incidents[%d].title", i), inc.Title)
			logger.DetectAndLog(a.sanitizer, entity.URN, fmt.Sprintf("incidents[%d].description", i), inc.Description)
		}
	}
}

// convertOwners converts DataHub owners to semantic owners.
func convertOwners(owners []types.Owner) []semantic.Owner {
	result := make([]semantic.Owner, len(owners))
	for i, owner := range owners {
		result[i] = semantic.Owner{
			URN:   owner.URN,
			Type:  ownerTypeToSemantic(owner.Type),
			Name:  owner.Name,
			Email: owner.Email,
		}
	}
	return result
}

// convertTags converts DataHub tags to string slice.
func convertTags(tags []types.Tag) []string {
	result := make([]string, len(tags))
	for i, tag := range tags {
		result[i] = tag.Name
	}
	return result
}

// convertGlossaryTerms converts DataHub glossary terms to semantic terms.
func convertGlossaryTerms(terms []types.GlossaryTerm) []semantic.GlossaryTerm {
	result := make([]semantic.GlossaryTerm, len(terms))
	for i, term := range terms {
		result[i] = semantic.GlossaryTerm{
			URN:         term.URN,
			Name:        term.Name,
			Description: term.Description,
		}
	}
	return result
}

// convertDomain converts DataHub domain to semantic domain.
func convertDomain(domain *types.Domain) *semantic.Domain {
	if domain == nil {
		return nil
	}
	return &semantic.Domain{
		URN:         domain.URN,
		Name:        domain.Name,
		Description: domain.Description,
	}
}

// convertDeprecation converts DataHub deprecation to semantic deprecation.
func convertDeprecation(dep *types.Deprecation) *semantic.Deprecation {
	if dep == nil {
		return nil
	}
	result := &semantic.Deprecation{
		Deprecated: dep.Deprecated, //nolint:staticcheck // Using deprecated field as it's the only option
		Note:       dep.Note,
		Actor:      dep.Actor,
	}
	if dep.DecommissionTime > 0 {
		t := time.Unix(dep.DecommissionTime/1000, 0)
		result.DecommDate = &t
	}
	return result
}

// convertProperties converts DataHub properties to string map.
func convertProperties(props map[string]any) map[string]string {
	if len(props) == 0 {
		return nil
	}
	result := make(map[string]string, len(props))
	for k, v := range props {
		if s, ok := v.(string); ok {
			result[k] = s
		}
	}
	return result
}

// convertTimestamp converts a millisecond timestamp to a time pointer.
func convertTimestamp(ms int64) *time.Time {
	if ms == 0 {
		return nil
	}
	t := time.Unix(ms/1000, 0)
	return &t
}

// fieldToColumnContext converts a DataHub schema field to semantic column context.
func (a *Adapter) fieldToColumnContext(field types.SchemaField) *semantic.ColumnContext {
	fieldName := extractFieldName(field.FieldPath)

	// Log any injection attempts in user-provided content
	logger := semantic.DefaultInjectionLogger
	logger.DetectAndLog(a.sanitizer, "column:"+fieldName, "description", field.Description)
	for i, term := range field.GlossaryTerms {
		logger.DetectAndLog(a.sanitizer, "column:"+fieldName, fmt.Sprintf("glossaryTerms[%d].description", i), term.Description)
	}

	cc := &semantic.ColumnContext{
		Name:        fieldName,
		Description: field.Description,
	}

	// Check for PII and sensitivity tags
	for _, tag := range field.Tags {
		tagLower := strings.ToLower(tag.Name)
		if strings.Contains(tagLower, "pii") {
			cc.IsPII = true
		}
		if strings.Contains(tagLower, "sensitive") || strings.Contains(tagLower, "confidential") {
			cc.IsSensitive = true
		}
		cc.Tags = append(cc.Tags, tag.Name)
	}

	// Convert glossary terms
	cc.GlossaryTerms = make([]semantic.GlossaryTerm, len(field.GlossaryTerms))
	for i, term := range field.GlossaryTerms {
		cc.GlossaryTerms[i] = semantic.GlossaryTerm{
			URN:         term.URN,
			Name:        term.Name,
			Description: term.Description,
		}
	}

	return cc
}

// lineageResultToInfo converts a DataHub lineage result to semantic lineage info.
func (*Adapter) lineageResultToInfo(result *types.LineageResult, direction semantic.LineageDirection, maxDepth int) *semantic.LineageInfo {
	info := &semantic.LineageInfo{
		Direction: direction,
		MaxDepth:  maxDepth,
		Entities:  make([]semantic.LineageEntity, len(result.Nodes)),
	}

	// Build edge map for quick lookup
	edgeMap := make(map[string][]string)
	for _, edge := range result.Edges {
		edgeMap[edge.Target] = append(edgeMap[edge.Target], edge.Source)
	}

	for i, node := range result.Nodes {
		entity := semantic.LineageEntity{
			URN:      node.URN,
			Type:     node.Type,
			Name:     node.Name,
			Platform: node.Platform,
			Depth:    node.Level,
		}

		// Add parent edges
		if parents, ok := edgeMap[node.URN]; ok {
			entity.Parents = make([]semantic.LineageEdge, len(parents))
			for j, parent := range parents {
				entity.Parents[j] = semantic.LineageEdge{URN: parent}
			}
		}

		info.Entities[i] = entity
	}

	return info
}

// ownerTypeToSemantic converts DataHub ownership type to semantic type.
// Currently all DataHub ownership types map to OwnerTypeUser; extend this
// mapping when additional semantic owner types are introduced.
func ownerTypeToSemantic(_ types.OwnershipType) semantic.OwnerType {
	return semantic.OwnerTypeUser
}

// extractFieldName extracts the simple field name from a FieldPath.
// FieldPath can be "user.address.city" - we want "city".
func extractFieldName(fieldPath string) string {
	parts := strings.Split(fieldPath, ".")
	if len(parts) == 0 {
		return fieldPath
	}
	return parts[len(parts)-1]
}

// convertStructuredProperties converts DataHub structured properties to semantic types.
func convertStructuredProperties(props []types.StructuredPropertyValue) []semantic.StructuredProperty {
	if len(props) == 0 {
		return nil
	}
	result := make([]semantic.StructuredProperty, len(props))
	for i, p := range props {
		sp := semantic.StructuredProperty{
			Values: p.Values,
		}
		if p.Definition != nil {
			sp.QualifiedName = p.Definition.QualifiedName
			sp.DisplayName = p.Definition.DisplayName
		}
		result[i] = sp
	}
	return result
}

// convertIncidents converts DataHub incident result to semantic types.
// Returns (total, incidents) where total comes from ir.Total and may exceed
// len(incidents) if DataHub paginates or truncates the incident list.
func convertIncidents(ir *types.IncidentResult) (int, []semantic.Incident) {
	if ir == nil || ir.Total == 0 {
		return 0, nil
	}
	incidents := make([]semantic.Incident, len(ir.Incidents))
	for i, inc := range ir.Incidents {
		incidents[i] = semantic.Incident{
			URN:         inc.URN,
			Type:        inc.Type,
			Title:       inc.Title,
			Description: inc.Description,
			State:       inc.State,
			Created:     inc.Created,
		}
	}
	return ir.Total, incidents
}

// convertDataContract converts DataHub data contract to semantic types.
func convertDataContract(dc *types.DataContract) *semantic.DataContractStatus {
	if dc == nil {
		return nil
	}
	result := &semantic.DataContractStatus{
		Status: dc.Status,
	}
	if len(dc.AssertionResults) > 0 {
		result.AssertionResults = make([]semantic.AssertionResult, len(dc.AssertionResults))
		for i, ar := range dc.AssertionResults {
			result.AssertionResults[i] = semantic.AssertionResult{
				AssertionURN: ar.AssertionURN,
				Type:         ar.Type,
			}
		}
	}
	return result
}

// Verify interface compliance.
var (
	_ semantic.Provider    = (*Adapter)(nil)
	_ semantic.URNResolver = (*Adapter)(nil)
)
