package userbehavior

import (
	"context"
	"strconv"
	"time"

	ubent "github.com/tidbcloud/devops/entity/userbehavior"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/clients/dataflowcli"
	"github.com/tidbcloud/devops/pkg/service/userbehavior/repo"
)

// Current only support scan by tenant
func (s *svc) ScanDMJobEvent(ctx context.Context) error {
	dmJobsubs, err := s.repo.QuerySubscriptions(ctx, repo.SubscriptionOpts.IsUBEvent(ubent.UBEventDMJob))
	if err != nil {
		return err
	}

	for _, sub := range dmJobsubs {
		// Current only support scan by tenant
		if len(sub.Selector.TenantSelectors) == 0 {
			continue
		}

		tenantId, err := strconv.ParseUint(string(sub.Selector.TenantSelectors[0]), 10, 64)
		if err != nil {
			s.logger.Info("cannot parse tenant id", "sub_id", sub.ID, "tenant_id", tenantId, "error", err)
			continue
		}

		includeDevTier := false
		clusters, _, err := s.coresvcCli.ListClusters(ctx, coresvccli.ListClustersParam{
			TenantIDs: []uint64{tenantId},
			IsDevTier: &includeDevTier,
		})
		if err != nil {
			s.logger.Info("cannot get clusters", "sub_id", sub.ID, "tenant_id", tenantId, "error", err)
			continue
		}

		for _, cluster := range clusters {
			jobs, _, err := s.dataflowCli.ListDMJobs(ctx, dataflowcli.ListDMJobsParam{
				OrgId:     tenantId,
				ClusterId: cluster.Id,
			})
			if err != nil {
				s.logger.Info("cannot get DM Jobs", "sub_id", sub.ID, "tenant_id", tenantId, "error", err)
			}

			for i := range jobs {
				createTime, err := time.Parse(time.RFC3339, jobs[i].CreateTime)
				if err != nil {
					s.logger.Info("parse create time failed", "create_time", jobs[i].CreateTime, "job_id", jobs[i].Id, "error", err)
				}
				if time.Now().Add(-15 * time.Minute).Before(createTime) {
					ev := newDMJobEvent(jobs[i], tenantId, cluster.Id)
					if ev != nil {
						s.ubCh <- ev
					}
				}
			}
		}

	}

	return nil
}