Subversion Repositories SmartDukaan

package com.spice.profitmandi.dao.service;

import org.hibernate.Session;

import org.hibernate.SessionFactory;

import org.hibernate.query.NativeQuery;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import java.util.*;

/**

 * Period analytics for the Beat Journey console — read-only native SQL, scoped to

 * the caller's downline via the auth-id / dtr-id sets, with {@code IN (:ids)} scope

 * filters and blocks for KPIs, funnel, scorecard, coverage, utilization, deferrals,

 * travel, outcomes, and findings.

 *

 * Empty scope is handled by substituting a sentinel id (-1) so every {@code IN}

 * clause is valid and simply returns zeros rather than throwing.

 */

@Service

public class BeatJourneyServiceImpl implements BeatJourneyService {

    @Autowired

    private SessionFactory sessionFactory;

    @Override

    public Map<String, Object> periodAnalytics(String start, String end, String today,

                                               List<Integer> authIds, List<Integer> dtrIds) {

        Session s = sessionFactory.getCurrentSession();

        List<Integer> auth = safeIds(authIds);

        List<Integer> dtr = safeIds(dtrIds);

        Map<String, Object> out = new LinkedHashMap<>();

        Map<String, Object> range = new LinkedHashMap<>();

        range.put("start", start);

        range.put("end", end);

        range.put("today", today);

        out.put("range", range);

        out.put("scopeCount", authIds == null ? 0 : authIds.size());

        List<Map<String, Object>> coverage = coverageByLevel(s, start, end, auth);

        Map<String, Object> funnel = funnel(s, start, end, auth, dtr);

        Map<String, Object> deferral = deferral(s, start, end, auth);

        Map<String, Object> utilization = utilization(s, start, end, dtr);

        Map<String, Object> travel = travel(s, start, end, auth, dtr);

        Map<String, Object> outcomes = outcomes(s, start, end, auth);

        List<Map<String, Object>> scorecard = scorecard(s, start, end, auth, dtr);

        Map<String, Object> kpis = kpis(s, start, end, auth, dtr, coverage, funnel, deferral);

        out.put("kpis", kpis);

        out.put("funnel", funnel);

        out.put("coverageByLevel", coverage);

        out.put("scorecard", scorecard);

        out.put("utilization", utilization);

        out.put("deferral", deferral);

        out.put("travel", travel);

        out.put("outcomes", outcomes);

        out.put("findings", findings(kpis, funnel, deferral, utilization, travel, outcomes, coverage, scorecard));

        // drill-down lists

        out.put("coverageDetail", coverageDetail(s, start, end, auth));

        out.put("deferralsList", deferralsList(s, start, end, auth));

        out.put("leadsList", leadsList(s, start, end, auth));

        return out;

    }

    // ---- KPIs ---------------------------------------------------------------

    private Map<String, Object> kpis(Session s, String ms, String me, List<Integer> auth, List<Integer> dtr,

                                     List<Map<String, Object>> coverage, Map<String, Object> funnel, Map<String, Object> deferral) {

        Map<String, Object> m = new LinkedHashMap<>();

        long planned = asLong(funnel.get("plannedStops"));

        long done = asLong(funnel.get("checkedOut"));

        m.put("plannedStops", planned);

        m.put("doneStops", done);

        m.put("adherencePct", planned == 0 ? 0 : Math.round(100.0 * done / planned));

        // L1 coverage row

        long l1With = 0, l1Total = 0;

        for (Map<String, Object> c : coverage) {

            if ("L1".equalsIgnoreCase(asStr(c.get("level")))) {

                l1With = asLong(c.get("withPjp"));

                l1Total = asLong(c.get("totalUsers"));

            }

        }

        m.put("coverageL1With", l1With);

        m.put("coverageL1Total", l1Total);

        m.put("coverageL1Pct", l1Total == 0 ? 0 : Math.round(100.0 * l1With / l1Total));

        // avg visits / active (punched-in) day, and total punch-in days (journeys)

        // A "punch-in day" is the attendance row (task_id=0) with a real check-in

        // time, matching the Today lens. (Literal mark_type='PUNCHIN' rows barely

        // exist in the data, so relying on them under-counts drastically.)

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT COUNT(DISTINCT CONCAT(user_id,'-',task_date)) FROM auth.location_tracking" +

            " WHERE task_date>=:ms AND task_date<=:me AND user_id IN (:dtr)" +

            "   AND (mark_type='PUNCHIN' OR (task_id=0 AND check_in_time IS NOT NULL AND check_in_time<>'00:00:00'))");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("dtr", dtr);

        long journeys = asLong(q.uniqueResult());

        m.put("journeys", journeys);

        m.put("avgVisitsPerActiveDay", journeys == 0 ? 0 : round1((double) asLong(funnel.get("checkedIn")) / journeys));

        // avg in-store discussion minutes

        NativeQuery<?> dq = s.createNativeQuery(

            "SELECT ROUND(AVG(NULLIF(TIME_TO_SEC(time_spent),0))/60,0) FROM auth.location_tracking" +

            " WHERE task_date>=:ms AND task_date<=:me AND task_id<>0 AND mark_type LIKE '%CHECKOUT%' AND user_id IN (:dtr)");

        dq.setParameter("ms", ms).setParameter("me", me).setParameterList("dtr", dtr);

        m.put("avgDiscussionMin", asLong(dq.uniqueResult()));

        m.put("autoMissed", asLong(deferral.get("total")));

        m.put("autoMissedSystemPct", asLong(deferral.get("systemPct")));

        // leads created on route

        NativeQuery<?> lq = s.createNativeQuery(

            "SELECT COUNT(*) FROM `user`.lead WHERE created_timestamp>=:ms AND created_timestamp<=CONCAT(:me,' 23:59:59') AND auth_id IN (:auth)");

        lq.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        m.put("leadsCreated", asLong(lq.uniqueResult()));

        return m;

    }

    // ---- Funnel -------------------------------------------------------------

    private Map<String, Object> funnel(Session s, String ms, String me, List<Integer> auth, List<Integer> dtr) {

        Map<String, Object> m = new LinkedHashMap<>();

        // Planned stops = partner stops (beat_route) + lead stops (lead_route, APPROVED)

        // on each scheduled beat-day, over active beats — exactly how the Today lens

        // (getAllScheduledBeats) counts, so Period reconciles with the day views.

        long planned = plannedCount(s, ms, me, auth, dtr, false, false)

                     + plannedCount(s, ms, me, auth, dtr, true, false);

        // ...of those, the planned stops that fall on a day the owner punched in

        long punchedIn = plannedCount(s, ms, me, auth, dtr, false, true)

                       + plannedCount(s, ms, me, auth, dtr, true, true);

        long checkedIn = countVisits(s, ms, me, dtr, "mark_type LIKE 'CHECKIN%'");

        long checkedOut = countVisits(s, ms, me, dtr, "(mark_type LIKE '%CHECKOUT%')");

        m.put("plannedStops", planned);

        m.put("punchedIn", punchedIn);

        m.put("checkedIn", checkedIn);

        m.put("checkedOut", checkedOut);

        m.put("dropPunch", planned == 0 ? 0 : Math.round(100.0 * (planned - punchedIn) / planned));

        m.put("dropCheckin", punchedIn == 0 ? 0 : Math.round(100.0 * (punchedIn - checkedIn) / punchedIn));

        m.put("dropCheckout", checkedIn == 0 ? 0 : Math.round(100.0 * (checkedIn - checkedOut) / checkedIn));

        return m;

    }

    // Planned stop count over scheduled beat-days of active beats. lead=false counts

    // partner stops (beat_route by day_number); lead=true counts approved lead stops

    // (lead_route by schedule_date). onlyPunchInDays restricts to days the owner

    // actually punched in (the funnel's "planned on punch-in days" stage).

    private long plannedCount(Session s, String ms, String me, List<Integer> auth,

                              List<Integer> dtr, boolean lead, boolean onlyPunchInDays) {

        String stopJoin = lead

            ? " JOIN `user`.lead_route x ON x.beat_id=sc.beat_id AND x.schedule_date=sc.start_date AND x.status='APPROVED'"

            : " JOIN `user`.beat_route x ON x.beat_id=sc.beat_id AND x.day_number=sc.day_number";

        String punchJoin = onlyPunchInDays

            ? " JOIN auth.auth_user au ON au.id=b.auth_user_id" +

              " JOIN dtr.users du ON du.email=au.email_id" +

              " JOIN (SELECT DISTINCT user_id, task_date FROM auth.location_tracking" +

              "        WHERE task_date>=:ms AND task_date<=:me AND user_id IN (:dtr)" +

              "          AND (mark_type='PUNCHIN' OR (task_id=0 AND check_in_time IS NOT NULL AND check_in_time<>'00:00:00'))) p" +

              "   ON p.user_id=du.id AND p.task_date=sc.start_date"

            : "";

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT COUNT(*) FROM `user`.beat_schedule sc" +

            " JOIN `user`.beat b ON b.id=sc.beat_id AND b.active=1 AND b.auth_user_id IN (:auth)" +

            stopJoin + punchJoin +

            " WHERE sc.start_date>=:ms AND sc.start_date<=:me");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        if (onlyPunchInDays) q.setParameterList("dtr", dtr);

        return asLong(q.uniqueResult());

    }

    // Counts visits across ALL task types (franchisee + lead + office) — task_id<>0

    // excludes the attendance/punch row — matching the Today lens, which tallies

    // every check-in/out regardless of type.

    private long countVisits(Session s, String ms, String me, List<Integer> dtr, String markPredicate) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT COUNT(*) FROM auth.location_tracking" +

            " WHERE task_date>=:ms AND task_date<=:me AND task_id<>0 AND " + markPredicate +

            " AND user_id IN (:dtr)");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("dtr", dtr);

        return asLong(q.uniqueResult());

    }

    // ---- Coverage by level --------------------------------------------------

    private List<Map<String, Object>> coverageByLevel(Session s, String ms, String me, List<Integer> auth) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT p.escalation_type lvl, COUNT(DISTINCT p.auth_user_id) total_users," +

            " COUNT(DISTINCT CASE WHEN sc.id IS NOT NULL THEN p.auth_user_id END) with_pjp" +

            " FROM cs.position p" +

            " LEFT JOIN `user`.beat b ON b.auth_user_id=p.auth_user_id AND b.active=1" +

            " LEFT JOIN `user`.beat_schedule sc ON sc.beat_id=b.id AND sc.start_date>=:ms AND sc.start_date<=:me" +

            " WHERE p.category_id=4 AND p.auth_user_id IN (:auth) GROUP BY p.escalation_type ORDER BY p.escalation_type");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        List<Map<String, Object>> list = new ArrayList<>();

        for (Object o : q.getResultList()) {

            Object[] r = (Object[]) o;

            long total = asLong(r[1]), with = asLong(r[2]);

            Map<String, Object> m = new LinkedHashMap<>();

            m.put("level", asStr(r[0]));

            m.put("totalUsers", total);

            m.put("withPjp", with);

            m.put("coveragePct", total == 0 ? 0 : Math.round(100.0 * with / total));

            list.add(m);

        }

        return list;

    }

    // ---- Per-executive scorecard -------------------------------------------

    private List<Map<String, Object>> scorecard(Session s, String ms, String me, List<Integer> auth, List<Integer> dtr) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT au.id uid, TRIM(CONCAT(au.first_name,' ',COALESCE(au.last_name,''))) nm, lvl.lvl," +

            " COALESCE(pl.planned,0)+COALESCE(plL.planned,0) planned, COALESCE(ac.done,0) done," +

            " COALESCE(ds.disc_min,0) disc_min, COALESCE(wk.work_sec,0) work_sec," +

            " COALESCE(ld.leads,0) leads, du.id duid" +

            " FROM (SELECT DISTINCT auth_user_id FROM `user`.beat WHERE active=1 AND auth_user_id IN (:auth)) o" +

            " JOIN auth.auth_user au ON au.id=o.auth_user_id" +

            " LEFT JOIN dtr.users du ON du.email=au.email_id" +

            " LEFT JOIN (SELECT auth_user_id, MIN(escalation_type) lvl FROM cs.position WHERE category_id=4 GROUP BY auth_user_id) lvl ON lvl.auth_user_id=au.id" +

            " LEFT JOIN (SELECT b.auth_user_id uid, COUNT(*) planned FROM `user`.beat_schedule sc" +

            "   JOIN `user`.beat b ON b.id=sc.beat_id AND b.active=1 JOIN `user`.beat_route r ON r.beat_id=sc.beat_id AND r.day_number=sc.day_number" +

            "   WHERE sc.start_date>=:ms AND sc.start_date<=:me GROUP BY b.auth_user_id) pl ON pl.uid=au.id" +

            " LEFT JOIN (SELECT b.auth_user_id uid, COUNT(*) planned FROM `user`.beat_schedule sc" +

            "   JOIN `user`.beat b ON b.id=sc.beat_id AND b.active=1 JOIN `user`.lead_route lr ON lr.beat_id=sc.beat_id AND lr.schedule_date=sc.start_date AND lr.status='APPROVED'" +

            "   WHERE sc.start_date>=:ms AND sc.start_date<=:me GROUP BY b.auth_user_id) plL ON plL.uid=au.id" +

            " LEFT JOIN (SELECT user_id uid, COUNT(*) done FROM auth.location_tracking" +

            "   WHERE task_date>=:ms AND task_date<=:me AND task_id<>0 AND mark_type LIKE 'CHECKIN%' GROUP BY user_id) ac ON ac.uid=du.id" +

            " LEFT JOIN (SELECT user_id uid, ROUND(AVG(NULLIF(TIME_TO_SEC(time_spent),0))/60,0) disc_min FROM auth.location_tracking" +

            "   WHERE task_date>=:ms AND task_date<=:me AND task_id<>0 AND mark_type LIKE '%CHECKOUT%' GROUP BY user_id) ds ON ds.uid=du.id" +

            " LEFT JOIN (SELECT user_id uid, SUM(GREATEST(TIME_TO_SEC(check_out_time)-TIME_TO_SEC(check_in_time),0)) work_sec FROM auth.location_tracking" +

            "   WHERE task_date>=:ms AND task_date<=:me AND task_id=0 AND check_out_time IS NOT NULL AND check_out_time<>'00:00:00' AND check_in_time IS NOT NULL AND check_in_time<>'00:00:00' GROUP BY user_id) wk ON wk.uid=du.id" +

            " LEFT JOIN (SELECT auth_id uid, COUNT(*) leads FROM `user`.lead WHERE created_timestamp>=:ms AND created_timestamp<=CONCAT(:me,' 23:59:59') GROUP BY auth_id) ld ON ld.uid=au.id" +

            " ORDER BY done DESC, planned DESC");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        // Geofence-offsite check-ins per user are computed in Java (not SQL): the

        // coords are stored as "lat,lng" strings, and the inline parsing/haversine

        // expression trips Hibernate's native-query parameter parser.

        Map<Integer, Integer> geoByUser = geoOffsiteByUser(s, ms, me, dtr);

        List<Map<String, Object>> list = new ArrayList<>();

        for (Object o : q.getResultList()) {

            Object[] r = (Object[]) o;

            long planned = asLong(r[3]), done = asLong(r[4]);

            Integer duid = (r[8] == null) ? null : ((Number) r[8]).intValue();

            Map<String, Object> m = new LinkedHashMap<>();

            m.put("authUserId", asLong(r[0]));

            m.put("userId", duid == null ? 0 : duid);

            m.put("name", asStr(r[1]));

            m.put("level", r[2] == null ? "-" : asStr(r[2]));

            m.put("planned", planned);

            m.put("done", done);

            m.put("adherencePct", planned == 0 ? 0 : Math.round(100.0 * done / planned));

            m.put("discussionMin", asLong(r[5]));

            m.put("workingHrs", round1(asDouble(r[6]) / 3600.0));

            m.put("leads", asLong(r[7]));

            m.put("geoFlags", duid == null ? 0 : geoByUser.getOrDefault(duid, 0));

            list.add(m);

        }

        return list;

    }

    // Per dtr-user count of check-ins recorded >50 m from the partner's saved

    // location. Pulls the raw "lat,lng" strings and does the haversine in Java.

    private Map<Integer, Integer> geoOffsiteByUser(Session s, String ms, String me, List<Integer> dtr) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT user_id, checkin_lat_lng, visit_location FROM auth.location_tracking" +

            " WHERE task_date>=:ms AND task_date<=:me AND task_type='franchisee-visit' AND mark_type LIKE 'CHECKIN%'" +

            " AND checkin_lat_lng LIKE '%,%' AND visit_location LIKE '%,%' AND user_id IN (:dtr)");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("dtr", dtr);

        Map<Integer, Integer> out = new HashMap<>();

        for (Object o : q.getResultList()) {

            Object[] r = (Object[]) o;

            if (r[0] == null) continue;

            double[] cin = parseLatLng(asStr(r[1]));

            double[] vis = parseLatLng(asStr(r[2]));

            if (cin == null || vis == null) continue;

            if (haversineMeters(cin[0], cin[1], vis[0], vis[1]) > 50.0) {

                int uid = ((Number) r[0]).intValue();

                out.merge(uid, 1, Integer::sum);

            }

        }

        return out;

    }

    private static double[] parseLatLng(String s) {

        if (s == null || !s.contains(",")) return null;

        String[] parts = s.split(",");

        if (parts.length != 2) return null;

        try {

            double a = Double.parseDouble(parts[0].trim());

            double b = Double.parseDouble(parts[1].trim());

            if (a == 0 && b == 0) return null;

            return new double[]{a, b};

        } catch (NumberFormatException e) {

            return null;

        }

    }

    private static double haversineMeters(double lat1, double lng1, double lat2, double lng2) {

        double R = 6371000d;

        double dLat = Math.toRadians(lat2 - lat1);

        double dLng = Math.toRadians(lng2 - lng1);

        double a = Math.sin(dLat / 2) * Math.sin(dLat / 2)

                + Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2))

                * Math.sin(dLng / 2) * Math.sin(dLng / 2);

        return 2 * R * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));

    }

    // ---- Working-hour utilization ------------------------------------------

    private Map<String, Object> utilization(Session s, String ms, String me, List<Integer> dtr) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT" +

            " (SELECT COALESCE(SUM(TIME_TO_SEC(time_spent)),0) FROM auth.location_tracking WHERE task_date>=:ms AND task_date<=:me AND task_id<>0 AND user_id IN (:dtr)) instore," +

            " (SELECT COALESCE(SUM(TIME_TO_SEC(transit_time)),0) FROM auth.location_tracking WHERE task_date>=:ms AND task_date<=:me AND task_id<>0 AND user_id IN (:dtr)) travel," +

            " (SELECT COALESCE(SUM(GREATEST(TIME_TO_SEC(check_out_time)-TIME_TO_SEC(check_in_time),0)),0) FROM auth.location_tracking WHERE task_date>=:ms AND task_date<=:me AND task_id=0 AND check_out_time IS NOT NULL AND check_out_time<>'00:00:00' AND check_in_time IS NOT NULL AND check_in_time<>'00:00:00' AND user_id IN (:dtr)) working");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("dtr", dtr);

        Object[] r = (Object[]) q.uniqueResult();

        double instore = asDouble(r[0]), travel = asDouble(r[1]), working = asDouble(r[2]);

        if (working < instore + travel) working = instore + travel; // guard: working must cover its parts

        double idle = Math.max(working - instore - travel, 0);

        Map<String, Object> m = new LinkedHashMap<>();

        m.put("inStoreHrs", round1(instore / 3600.0));

        m.put("travelHrs", round1(travel / 3600.0));

        m.put("idleHrs", round1(idle / 3600.0));

        m.put("workingHrs", round1(working / 3600.0));

        m.put("inStorePct", working == 0 ? 0 : (int) Math.round(100.0 * instore / working));

        m.put("travelPct", working == 0 ? 0 : (int) Math.round(100.0 * travel / working));

        m.put("idlePct", working == 0 ? 0 : (int) Math.round(100.0 * idle / working));

        return m;

    }

    // ---- Deferrals & SLA ----------------------------------------------------

    private Map<String, Object> deferral(Session s, String ms, String me, List<Integer> auth) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT COUNT(*) total," +

            " SUM(status='RESCHEDULED') rescheduled, SUM(status='CANCELLED') cancelled," +

            " SUM(action_by IS NOT NULL) actioned," +

            " SUM(reason LIKE 'Beat missed%' OR reason LIKE 'Not marked%') system_gen," +

            " ROUND(AVG(CASE WHEN action_by IS NOT NULL THEN TIMESTAMPDIFF(HOUR, created_timestamp, updated_timestamp)/24.0 END),1) avg_days" +

            " FROM `user`.beat_deferred_visit WHERE deferred_date>=:ms AND deferred_date<=:me AND auth_user_id IN (:auth)");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        Object[] r = (Object[]) q.uniqueResult();

        long total = asLong(r[0]), system = asLong(r[4]);

        long human = total - system;

        Map<String, Object> m = new LinkedHashMap<>();

        m.put("total", total);

        m.put("rescheduled", asLong(r[1]));

        m.put("cancelled", asLong(r[2]));

        m.put("actioned", asLong(r[3]));

        m.put("systemGenerated", system);

        m.put("human", human);

        m.put("avgDaysToAction", asDouble(r[5]));

        m.put("systemPct", total == 0 ? 0 : Math.round(100.0 * system / total));

        m.put("humanPct", total == 0 ? 0 : Math.round(100.0 * human / total));

        return m;

    }

    // ---- Drill-down lists ---------------------------------------------------

    // Per-executive PJP coverage (with/without a plan in range) — powers the

    // PJP coverage KPI and the coverage-by-level bar drill-downs.

    private List<Map<String, Object>> coverageDetail(Session s, String ms, String me, List<Integer> auth) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT au.id uid, TRIM(CONCAT(au.first_name,' ',COALESCE(au.last_name,''))) nm, p.lvl," +

            " MAX(CASE WHEN sc.id IS NOT NULL THEN 1 ELSE 0 END) has_plan" +

            " FROM (SELECT auth_user_id, MIN(escalation_type) lvl FROM cs.position WHERE category_id=4 AND auth_user_id IN (:auth) GROUP BY auth_user_id) p" +

            " JOIN auth.auth_user au ON au.id=p.auth_user_id" +

            " LEFT JOIN `user`.beat b ON b.auth_user_id=p.auth_user_id AND b.active=1" +

            " LEFT JOIN `user`.beat_schedule sc ON sc.beat_id=b.id AND sc.start_date>=:ms AND sc.start_date<=:me" +

            " GROUP BY au.id, nm, p.lvl ORDER BY p.lvl, nm");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        List<Map<String, Object>> list = new ArrayList<>();

        for (Object o : q.getResultList()) {

            Object[] r = (Object[]) o;

            Map<String, Object> m = new LinkedHashMap<>();

            m.put("authUserId", asLong(r[0]));

            m.put("name", asStr(r[1]));

            m.put("level", asStr(r[2]));

            m.put("hasPlan", asLong(r[3]) > 0);

            list.add(m);

        }

        return list;

    }

    // Every deferral in range (exec, partner, reason, date, status, system flag).

    private List<Map<String, Object>> deferralsList(Session s, String ms, String me, List<Integer> auth) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT TRIM(CONCAT(au.first_name,' ',COALESCE(au.last_name,''))) nm, d.display_name, d.reason," +

            " d.deferred_date, d.status, d.task_type," +

            " (d.reason LIKE 'Beat missed%' OR d.reason LIKE 'Not marked%') sys" +

            " FROM `user`.beat_deferred_visit d JOIN auth.auth_user au ON au.id=d.auth_user_id" +

            " WHERE d.deferred_date>=:ms AND d.deferred_date<=:me AND d.auth_user_id IN (:auth)" +

            " ORDER BY d.deferred_date, nm");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        List<Map<String, Object>> list = new ArrayList<>();

        for (Object o : q.getResultList()) {

            Object[] r = (Object[]) o;

            Map<String, Object> m = new LinkedHashMap<>();

            m.put("exec", asStr(r[0]));

            m.put("partner", asStr(r[1]));

            m.put("reason", asStr(r[2]));

            m.put("date", asStr(r[3]));

            m.put("status", asStr(r[4]));

            m.put("taskType", asStr(r[5]));

            m.put("system", asLong(r[6]) > 0);

            list.add(m);

        }

        return list;

    }

    // Every lead created in range (exec, outlet, city, date, status).

    private List<Map<String, Object>> leadsList(Session s, String ms, String me, List<Integer> auth) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT TRIM(CONCAT(au.first_name,' ',COALESCE(au.last_name,''))) nm," +

            " COALESCE(NULLIF(l.outlet_name,''), TRIM(CONCAT(COALESCE(l.first_name,''),' ',COALESCE(l.last_name,'')))) lead," +

            " l.city, DATE(l.created_timestamp) dt, l.status" +

            " FROM `user`.lead l JOIN auth.auth_user au ON au.id=l.auth_id" +

            " WHERE l.created_timestamp>=:ms AND l.created_timestamp<=CONCAT(:me,' 23:59:59') AND l.auth_id IN (:auth)" +

            " ORDER BY dt, nm");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        List<Map<String, Object>> list = new ArrayList<>();

        for (Object o : q.getResultList()) {

            Object[] r = (Object[]) o;

            Map<String, Object> m = new LinkedHashMap<>();

            m.put("exec", asStr(r[0]));

            m.put("lead", asStr(r[1]));

            m.put("city", asStr(r[2]));

            m.put("date", asStr(r[3]));

            m.put("status", asStr(r[4]));

            list.add(m);

        }

        return list;

    }

    // ---- Travel & route quality --------------------------------------------

    private Map<String, Object> travel(Session s, String ms, String me, List<Integer> auth, List<Integer> dtr) {

        Map<String, Object> m = new LinkedHashMap<>();

        String legFrom =

            " FROM `user`.beat_route a" +

            " JOIN `user`.beat bt ON bt.id=a.beat_id AND bt.auth_user_id IN (:auth)" +

            " JOIN `user`.beat_route b ON b.beat_id=a.beat_id AND b.day_number=a.day_number AND b.sequence_order=a.sequence_order+1 AND b.active=1" +

            " JOIN (SELECT id, CAST(latitude AS DECIMAL(10,7)) lat, CAST(longitude AS DECIMAL(10,7)) lng FROM fofo.fofo_store WHERE latitude REGEXP '^[0-9.]+$') al ON al.id=a.fofo_id" +

            " JOIN (SELECT id, CAST(latitude AS DECIMAL(10,7)) lat, CAST(longitude AS DECIMAL(10,7)) lng FROM fofo.fofo_store WHERE latitude REGEXP '^[0-9.]+$') bl ON bl.id=b.fofo_id" +

            " WHERE a.active=1";

        String legKm =

            "6371*2*ASIN(SQRT(POWER(SIN(RADIANS(bl.lat-al.lat)/2),2)+COS(RADIANS(al.lat))*COS(RADIANS(bl.lat))*POWER(SIN(RADIANS(bl.lng-al.lng)/2),2)))";

        NativeQuery<?> bq = s.createNativeQuery(

            "SELECT bucket, COUNT(*) n FROM (" +

            "  SELECT CASE WHEN km<2 THEN '0-2 km' WHEN km<5 THEN '2-5 km' WHEN km<15 THEN '5-15 km'" +

            "    WHEN km<50 THEN '15-50 km' ELSE '50 km+' END bucket," +

            "  CASE WHEN km<2 THEN 1 WHEN km<5 THEN 2 WHEN km<15 THEN 3 WHEN km<50 THEN 4 ELSE 5 END ord" +

            "  FROM (SELECT " + legKm + " km" + legFrom + ") l" +

            " ) z GROUP BY bucket, ord ORDER BY ord");

        bq.setParameterList("auth", auth);

        List<Map<String, Object>> buckets = new ArrayList<>();

        for (Object o : bq.getResultList()) {

            Object[] r = (Object[]) o;

            Map<String, Object> b = new LinkedHashMap<>();

            b.put("bucket", asStr(r[0]));

            b.put("count", asLong(r[1]));

            buckets.add(b);

        }

        m.put("legBuckets", buckets);

        NativeQuery<?> sq = s.createNativeQuery(

            "SELECT ROUND(AVG(km),1) avg_leg, ROUND(MAX(km),0) max_leg, SUM(km>15) over15, COUNT(*) total" +

            " FROM (SELECT " + legKm + " km" + legFrom + ") l");

        sq.setParameterList("auth", auth);

        Object[] sr = (Object[]) sq.uniqueResult();

        long totalLegs = asLong(sr[3]), over15 = asLong(sr[2]);

        m.put("avgLegKm", asDouble(sr[0]));

        m.put("maxLegKm", asDouble(sr[1]));

        m.put("pctOver15", totalLegs == 0 ? 0 : Math.round(100.0 * over15 / totalLegs));

        NativeQuery<?> tq = s.createNativeQuery(

            "SELECT ROUND(SUM(TIME_TO_SEC(transit_time))/3600,1) transit_hrs, ROUND(SUM(TIME_TO_SEC(estimated_time))/3600,1) est_hrs" +

            " FROM auth.location_tracking WHERE task_date>=:ms AND task_date<=:me AND task_type='franchisee-visit' AND user_id IN (:dtr)");

        tq.setParameter("ms", ms).setParameter("me", me).setParameterList("dtr", dtr);

        Object[] tr = (Object[]) tq.uniqueResult();

        double transit = asDouble(tr[0]), est = asDouble(tr[1]);

        m.put("actualVsEstimateRatio", est == 0 ? 0 : round1(transit / est));

        return m;

    }

    // ---- Outcomes & visit quality ------------------------------------------

    private Map<String, Object> outcomes(Session s, String ms, String me, List<Integer> auth) {

        NativeQuery<?> q = s.createNativeQuery(

            "SELECT COUNT(*) total," +

            " SUM(partner_remark IS NOT NULL AND TRIM(partner_remark)<>'') remark," +

            " ROUND(AVG(NULLIF(rbm_rating,0)),1) avg_rating," +

            " SUM(instore_visibility IS NOT NULL AND TRIM(instore_visibility)<>'') audit," +

            " SUM(franchise_activity_id>0) nextv" +

            " FROM `user`.my_franchisee_visit" +

            " WHERE auth_id IN (:auth) AND COALESCE(visit_timestamp, created_timestamp)>=:ms" +

            " AND COALESCE(visit_timestamp, created_timestamp)<=CONCAT(:me,' 23:59:59')");

        q.setParameter("ms", ms).setParameter("me", me).setParameterList("auth", auth);

        Object[] r = (Object[]) q.uniqueResult();

        long total = asLong(r[0]);

        Map<String, Object> m = new LinkedHashMap<>();

        m.put("completedVisits", total);

        m.put("remarkFilledPct", total == 0 ? 0 : Math.round(100.0 * asLong(r[1]) / total));

        m.put("avgRating", asDouble(r[2]));

        m.put("auditDonePct", total == 0 ? 0 : Math.round(100.0 * asLong(r[3]) / total));

        m.put("nextVisitPct", total == 0 ? 0 : Math.round(100.0 * asLong(r[4]) / total));

        return m;

    }

    // ---- "What to fix" findings (computed in Java) --------------------------

    private List<Map<String, Object>> findings(Map<String, Object> kpis, Map<String, Object> funnel,

                                               Map<String, Object> deferral, Map<String, Object> util,

                                               Map<String, Object> travel, Map<String, Object> outcomes,

                                               List<Map<String, Object>> coverage, List<Map<String, Object>> scorecard) {

        List<Map<String, Object>> f = new ArrayList<>();

        long avgDisc = asLong(kpis.get("avgDiscussionMin"));

        long rushed = scorecard.stream().filter(r -> asLong(r.get("discussionMin")) > 0 && asLong(r.get("discussionMin")) < 10).count();

        if (avgDisc > 0 && (avgDisc < 30 || rushed > 0)) {

            f.add(finding("high", "Rushed partner visits",

                "Avg discussion is " + avgDisc + " min vs a 30–45 min target" + (rushed > 0 ? ", and " + rushed + " executive(s) average under 10 min" : "") + " — too short for a real outlet conversation.",

                "Set a minimum on-store time and review visit quality."));

        }

        long geo = scorecard.stream().mapToLong(r -> asLong(r.get("geoFlags"))).sum();

        if (geo > 0) {

            f.add(finding("high", "Fake-visit risk",

                geo + " check-in(s) were >50 m from the store (check-in GPS vs the partner's saved location).",

                "Enforce geofenced check-in and flag these journeys for audit."));

        }

        for (Map<String, Object> c : coverage) {

            if ("L1".equalsIgnoreCase(asStr(c.get("level")))) {

                long total = asLong(c.get("totalUsers")), with = asLong(c.get("withPjp"));

                if (total - with > 0) {

                    f.add(finding("med", "Coverage gap",

                        (total - with) + " of " + total + " L1 executives have no monthly PJP — they run no planned beat this period.",

                        "Make a monthly PJP mandatory and approved before month start."));

                }

            }

        }

        long sysPct = asLong(deferral.get("systemPct"));

        double sla = asDouble(deferral.get("avgDaysToAction"));

        if (asLong(deferral.get("total")) > 0 && (sysPct >= 50 || sla >= 2)) {

            f.add(finding("med", "Deferral leakage",

                sysPct + "% of deferrals are system auto-flags" + (sla > 0 ? ", and genuine deferrals sit " + sla + " days before a head acts" : "") + ".",

                "Require a reason on real deferrals and add a manager-action SLA."));

        }

        long idlePct = asLong(util.get("idlePct"));

        if (idlePct >= 20) {

            f.add(finding("med", "A large part of the day is idle",

                idlePct + "% of the working day (" + util.get("idleHrs") + " h) is neither travel nor in-store.",

                "Investigate idle windows; tighten beat sequencing to cut dead time."));

        }

        long over15 = asLong(travel.get("pctOver15"));

        if (over15 >= 15) {

            f.add(finding("med", "Route inefficiency",

                over15 + "% of legs exceed 15 km — consecutive stops aren't geographically clustered.",

                "Re-sequence beats by proximity and cap daily travel."));

        }

        long journeys = asLong(kpis.get("journeys")), leads = asLong(kpis.get("leadsCreated"));

        if (journeys > 0 && leads < journeys * 0.25) {

            f.add(finding("low", "Lead capture is low",

                "Only " + leads + " leads across " + journeys + " journeys — executives aren't logging the new shops they pass.",

                "Add a per-beat lead nudge; recognise top loggers."));

        }

        return f;

    }

    private static Map<String, Object> finding(String sev, String title, String detail, String action) {

        Map<String, Object> m = new LinkedHashMap<>();

        m.put("severity", sev);

        m.put("title", title);

        m.put("detail", detail);

        m.put("action", action);

        return m;

    }

    // ---- helpers ------------------------------------------------------------

    private static List<Integer> safeIds(List<Integer> ids) {

        if (ids == null || ids.isEmpty()) return Collections.singletonList(-1);

        return ids;

    }

    private static long asLong(Object o) {

        return o == null ? 0L : ((Number) o).longValue();

    }

    private static double asDouble(Object o) {

        return o == null ? 0.0 : ((Number) o).doubleValue();

    }

    private static String asStr(Object o) {

        return o == null ? "" : o.toString();

    }

    private static double round1(double v) {

        return Math.round(v * 10.0) / 10.0;

    }

}