muyu-apiserver/gateway/internal/repo/crmrepo.go

package repo

import (
	"context"
	"database/sql"
	"errors"
	"time"

	"muyu-apiserver/pkg/uid"

	"github.com/zeromicro/go-zero/core/stores/sqlx"
)

type (
	CrmRepo struct {
		conn sqlx.SqlConn
	}

	RelationView struct {
		RelationshipId string       `db:"relationship_id" json:"relationshipId"`
		FromTenantId   string       `db:"from_tenant_id" json:"fromTenantId"`
		FromTenantName string       `db:"from_tenant_name" json:"fromTenantName"`
		ToTenantId     string       `db:"to_tenant_id" json:"toTenantId"`
		ToTenantName   string       `db:"to_tenant_name" json:"toTenantName"`
		RelationType   string       `db:"relation_type" json:"relationType"`
		Status         int64        `db:"status" json:"status"`
		ValidFrom      time.Time    `db:"valid_from" json:"validFrom"`
		ValidTo        sql.NullTime `db:"valid_to" json:"validTo,omitempty"`
	}

	CreateRelationInput struct {
		OwnerTenantId string
		FromTenantId  string
		ToTenantId    string
		RelationType  string
		Status        int64
		ValidFrom     time.Time
		ValidTo       sql.NullTime
		Source        string
		CreatedBy     string
	}

	UpdateRelationInput struct {
		RelationshipId string
		OwnerTenantId  string
		Status         int64
		ValidFrom      time.Time
		ValidTo        sql.NullTime
		UpdatedBy      string
	}

	RelationHistory struct {
		RelationshipId string         `db:"relationship_id" json:"relationshipId"`
		ChangeType     string         `db:"change_type" json:"changeType"`
		BeforeJSON     sql.NullString `db:"before_json" json:"beforeJson"`
		AfterJSON      sql.NullString `db:"after_json" json:"afterJson"`
		ChangedBy      string         `db:"changed_by" json:"changedBy"`
		ChangedAt      time.Time      `db:"changed_at" json:"changedAt"`
	}
)

func NewCrmRepo(conn sqlx.SqlConn) *CrmRepo {
	return &CrmRepo{conn: conn}
}

func (r *CrmRepo) SearchTenants(ctx context.Context, keyword string, limit int) ([]GraphNode, error) {
	if limit <= 0 || limit > 50 {
		limit = 20
	}
	nodes := make([]GraphNode, 0)
	if keyword == "" {
		if err := r.conn.QueryRowsCtx(ctx, &nodes,
			"SELECT tenant_id, tenant_name FROM crm_tenant WHERE status = 1 ORDER BY tenant_name LIMIT $1", limit); err != nil {
			return nil, err
		}
	} else {
		if err := r.conn.QueryRowsCtx(ctx, &nodes,
			"SELECT tenant_id, tenant_name FROM crm_tenant WHERE status = 1 AND tenant_name LIKE $1 ORDER BY tenant_name LIMIT $2",
			"%"+keyword+"%", limit); err != nil {
			return nil, err
		}
	}
	return nodes, nil
}

func (r *CrmRepo) AddUserToTenant(ctx context.Context, userId, tenantId string) error {
	_, err := r.conn.ExecCtx(ctx,
		`INSERT INTO crm_tenant_user (user_id, tenant_id, is_primary, status)
		 VALUES ($1, $2, 1, 1)
		 ON CONFLICT (user_id, tenant_id) DO NOTHING`,
		userId, tenantId,
	)
	return err
}

func (r *CrmRepo) VerifyUserTenant(ctx context.Context, userId, tenantId string) (bool, error) {
	var count int64
	err := r.conn.QueryRowCtx(ctx, &count,
		"SELECT COUNT(1) FROM crm_tenant_user WHERE user_id = $1 AND tenant_id = $2 AND status = 1",
		userId, tenantId,
	)
	if err != nil {
		return false, err
	}
	return count > 0, nil
}

func (r *CrmRepo) ResolvePrimaryTenant(ctx context.Context, userId string) (string, error) {
	var tenantId string
	err := r.conn.QueryRowCtx(ctx, &tenantId,
		"SELECT tenant_id FROM crm_tenant_user WHERE user_id = $1 AND status = 1 ORDER BY is_primary DESC, id ASC LIMIT 1",
		userId,
	)
	if err != nil {
		return "", err
	}
	return tenantId, nil
}

func (r *CrmRepo) ListUpstream(ctx context.Context, tenantId string) ([]RelationView, error) {
	rows := make([]RelationView, 0)
	query := `
SELECT r.relationship_id, r.from_tenant_id, ft.tenant_name AS from_tenant_name,
       r.to_tenant_id, tt.tenant_name AS to_tenant_name,
       r.relation_type, r.status, r.valid_from, r.valid_to
FROM crm_tenant_relationship r
JOIN crm_tenant ft ON ft.tenant_id = r.from_tenant_id
JOIN crm_tenant tt ON tt.tenant_id = r.to_tenant_id
WHERE r.owner_tenant_id = $1
  AND r.to_tenant_id = $2
  AND r.status = 1
  AND r.valid_from <= NOW()
  AND (r.valid_to IS NULL OR r.valid_to > NOW())
ORDER BY r.updated_at DESC`
	if err := r.conn.QueryRowsCtx(ctx, &rows, query, tenantId, tenantId); err != nil {
		return nil, err
	}
	return rows, nil
}

func (r *CrmRepo) ListDownstream(ctx context.Context, tenantId string) ([]RelationView, error) {
	rows := make([]RelationView, 0)
	query := `
SELECT r.relationship_id, r.from_tenant_id, ft.tenant_name AS from_tenant_name,
       r.to_tenant_id, tt.tenant_name AS to_tenant_name,
       r.relation_type, r.status, r.valid_from, r.valid_to
FROM crm_tenant_relationship r
JOIN crm_tenant ft ON ft.tenant_id = r.from_tenant_id
JOIN crm_tenant tt ON tt.tenant_id = r.to_tenant_id
WHERE r.owner_tenant_id = $1
  AND r.from_tenant_id = $2
  AND r.status = 1
  AND r.valid_from <= NOW()
  AND (r.valid_to IS NULL OR r.valid_to > NOW())
ORDER BY r.updated_at DESC`
	if err := r.conn.QueryRowsCtx(ctx, &rows, query, tenantId, tenantId); err != nil {
		return nil, err
	}
	return rows, nil
}

func (r *CrmRepo) CreateRelation(ctx context.Context, in CreateRelationInput) (string, error) {
	relID := "rel_" + uid.Generate()[:16]
	if in.Source == "" {
		in.Source = "manual"
	}
	if in.Status == 0 {
		in.Status = 1
	}
	_, err := r.conn.ExecCtx(ctx, `
INSERT INTO crm_tenant_relationship
  (relationship_id, owner_tenant_id, from_tenant_id, to_tenant_id, relation_type, status, valid_from, valid_to, source, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
		relID, in.OwnerTenantId, in.FromTenantId, in.ToTenantId, in.RelationType,
		in.Status, in.ValidFrom, in.ValidTo, in.Source, in.CreatedBy)
	if err != nil {
		return "", err
	}

	histID := "rh_" + uid.Generate()[:16]
	_, _ = r.conn.ExecCtx(ctx, `
INSERT INTO crm_tenant_relationship_history
  (history_id, relationship_id, change_type, before_json, after_json, changed_by)
VALUES ($1, $2, 'create', NULL, jsonb_build_object('status', $3::text, 'relationType', $4), $5)`,
		histID, relID, in.Status, in.RelationType, in.CreatedBy)

	return relID, nil
}

func (r *CrmRepo) UpdateRelation(ctx context.Context, in UpdateRelationInput) error {
	res, err := r.conn.ExecCtx(ctx, `
UPDATE crm_tenant_relationship
SET status = $1, valid_from = $2, valid_to = $3, updated_at = NOW()
WHERE relationship_id = $4 AND owner_tenant_id = $5`,
		in.Status, in.ValidFrom, in.ValidTo, in.RelationshipId, in.OwnerTenantId)
	if err != nil {
		return err
	}
	affected, _ := res.RowsAffected()
	if affected == 0 {
		return errors.New("relationship not found")
	}

	histID := "rh_" + uid.Generate()[:16]
	_, _ = r.conn.ExecCtx(ctx, `
INSERT INTO crm_tenant_relationship_history
  (history_id, relationship_id, change_type, before_json, after_json, changed_by)
VALUES ($1, $2, 'update', NULL, jsonb_build_object('status', $3::text), $4)`,
		histID, in.RelationshipId, in.Status, in.UpdatedBy)
	return nil
}

func (r *CrmRepo) ListHistory(ctx context.Context, ownerTenantId string, page, pageSize int64) ([]RelationHistory, int64, error) {
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 20
	}

	var total int64
	if err := r.conn.QueryRowCtx(ctx, &total, `
SELECT COUNT(1)
FROM crm_tenant_relationship_history h
JOIN crm_tenant_relationship r ON r.relationship_id = h.relationship_id
WHERE r.owner_tenant_id = $1`, ownerTenantId); err != nil {
		return nil, 0, err
	}

	offset := (page - 1) * pageSize
	rows := make([]RelationHistory, 0)
	if err := r.conn.QueryRowsCtx(ctx, &rows, `
SELECT h.relationship_id, h.change_type, h.before_json, h.after_json, h.changed_by, h.changed_at
FROM crm_tenant_relationship_history h
JOIN crm_tenant_relationship r ON r.relationship_id = h.relationship_id
WHERE r.owner_tenant_id = $1
ORDER BY h.changed_at DESC
LIMIT $2 OFFSET $3`, ownerTenantId, pageSize, offset); err != nil {
		return nil, 0, err
	}
	return rows, total, nil
}

type GraphNode struct {
	TenantId   string `db:"tenant_id" json:"id"`
	TenantName string `db:"tenant_name" json:"name"`
}

type GraphEdge struct {
	Source         string `db:"from_tenant_id" json:"source"`
	Target         string `db:"to_tenant_id" json:"target"`
	RelationType   string `db:"relation_type" json:"relationType"`
	RelationshipId string `db:"relationship_id" json:"relationshipId"`
}

type FullGraph struct {
	Nodes []GraphNode `json:"nodes"`
	Edges []GraphEdge `json:"edges"`
}

func (r *CrmRepo) GetFullGraph(ctx context.Context) (*FullGraph, error) {
	nodes := make([]GraphNode, 0)
	if err := r.conn.QueryRowsCtx(ctx, &nodes,
		"SELECT tenant_id, tenant_name FROM crm_tenant WHERE status = 1 ORDER BY tenant_id"); err != nil {
		return nil, err
	}

	edges := make([]GraphEdge, 0)
	if err := r.conn.QueryRowsCtx(ctx, &edges, `
SELECT DISTINCT r.relationship_id, r.from_tenant_id, r.to_tenant_id, r.relation_type
FROM crm_tenant_relationship r
WHERE r.status = 1
  AND r.valid_from <= NOW()
  AND (r.valid_to IS NULL OR r.valid_to > NOW())
ORDER BY r.from_tenant_id, r.to_tenant_id`); err != nil {
		return nil, err
	}

	return &FullGraph{Nodes: nodes, Edges: edges}, nil
}