1CREATE SCHEMA IF NOT EXISTS td;
 2
 3CREATE TABLE IF NOT EXISTS td.job_system (
 4    id              UUID        DEFAULT gen_random_uuid() PRIMARY KEY,
 5    job_name        VARCHAR(255) NOT NULL,
 6    cron_expression TEXT         NOT NULL,          -- thời điểm chạy
 7    assembly_file   TEXT         NOT NULL,          -- vd: MyApp.Jobs.dll
 8    namespace_class TEXT         NOT NULL,          -- vd: MyApp.Jobs.SendSummaryEmail
 9    is_active       BOOLEAN      DEFAULT TRUE,
10    created_at      TIMESTAMP    DEFAULT NOW()
11);

1CREATE TABLE IF NOT EXISTS td.job_user (
2    id              UUID        DEFAULT gen_random_uuid() PRIMARY KEY,
3    database_id     UUID        NOT NULL,   -- id database user, worker dùng để lấy connection string
4    cron_expression TEXT        NOT NULL,
5    job_type        VARCHAR(50) NOT NULL,   -- loại nghiệp vụ: vd 'send_email'
6    is_active       BOOLEAN     DEFAULT TRUE,
7    created_at      TIMESTAMP   DEFAULT NOW()
8);

 1DateTime currentTime = DateTime.Now;
 2
 3CronExpressionModel cronExp = new CronExpressionBuilder()
 4    .SetStartTimeAndEndtime(currentTime.AddDays(-30), currentTime)
 5    .SetMonthly(30)          // mỗi tháng, ngày 30
 6    .BuildHour(6)            // lúc 6h sáng
 7    .GetResult();
 8
 9string cronExpression = cronExp.ToString();
10// "* * 6 30|L * * * 2026-01-01 2026-08-01"

  1DROP FUNCTION IF EXISTS td.is_cron_expression_satisfied(cron_expression VARCHAR, target_datetime TIMESTAMP WITHOUT TIME ZONE);
  2CREATE OR REPLACE FUNCTION td.is_cron_expression_satisfied(
  3    cron_expression VARCHAR,
  4    target_datetime TIMESTAMP WITHOUT TIME ZONE
  5)
  6RETURNS BOOLEAN
  7AS $$
  8DECLARE
  9    cron_parts VARCHAR[] = STRING_TO_ARRAY(cron_expression, ' ');
 10    seconds VARCHAR;
 11    minutes VARCHAR;
 12    hours VARCHAR;
 13    day_of_month VARCHAR;
 14    month VARCHAR;
 15    day_of_week VARCHAR;
 16    year VARCHAR;
 17    start_time VARCHAR;
 18    end_time VARCHAR;
 19    target_second INT = EXTRACT(SECOND FROM target_datetime);
 20    target_minute INT = EXTRACT(MINUTE FROM target_datetime);
 21    target_hour INT = EXTRACT(HOUR FROM target_datetime);
 22    target_day_of_month INT = EXTRACT(DAY FROM target_datetime);
 23    target_month INT = EXTRACT(MONTH FROM target_datetime);
 24    target_day_of_week INT = EXTRACT(DOW FROM target_datetime) + 1;
 25    target_year INT = EXTRACT(YEAR FROM target_datetime);
 26BEGIN
 27    seconds       := coalesce(cron_parts[1], '*');
 28    minutes       := coalesce(cron_parts[2], '*');
 29    hours         := coalesce(cron_parts[3], '*');
 30    day_of_month  := coalesce(cron_parts[4], '*');
 31    month         := coalesce(cron_parts[5], '*');
 32    day_of_week   := coalesce(cron_parts[6], '*');
 33    year          := coalesce(cron_parts[7], '*');
 34    start_time    := coalesce(cron_parts[8], '*');
 35    end_time      := coalesce(cron_parts[9], '*');
 36
 37    RETURN (
 38        (
 39            (start_time = '*' AND end_time = '*')
 40            OR (target_datetime BETWEEN start_time::date AND end_time::date)
 41        ) AND
 42        td.is_cron_value_satisfied(seconds, target_second, target_datetime) AND
 43        td.is_cron_value_satisfied(minutes, target_minute, target_datetime) AND
 44        td.is_cron_value_satisfied(hours, target_hour, target_datetime) AND
 45        td.is_cron_value_satisfied(day_of_month, target_day_of_month, target_datetime) AND
 46        td.is_cron_value_satisfied(month, target_month, target_datetime) AND
 47        td.is_cron_value_satisfied(day_of_week, target_day_of_week, target_datetime) AND
 48        td.is_cron_value_satisfied(year, target_year, target_datetime)
 49    );
 50END;
 51$$ LANGUAGE plpgsql;
 52
 53DROP FUNCTION IF EXISTS td.is_cron_value_satisfied(cron_value VARCHAR, target_value INT, target_datetime TIMESTAMP WITHOUT TIME ZONE);
 54CREATE OR REPLACE FUNCTION td.is_cron_value_satisfied(
 55    cron_value VARCHAR,
 56    target_value INT,
 57    target_datetime TIMESTAMP WITHOUT TIME ZONE
 58)
 59RETURNS BOOLEAN
 60AS $$
 61DECLARE
 62    cron_parts VARCHAR[] = REGEXP_SPLIT_TO_ARRAY(cron_value, ',');
 63    i INT;
 64BEGIN
 65    FOR i IN 1..ARRAY_LENGTH(cron_parts, 1) LOOP
 66        IF cron_parts[i] = '*' THEN
 67            -- mọi giá trị đều phù hợp
 68            RETURN TRUE;
 69        ELSIF cron_parts[i] LIKE '%-%' THEN
 70            -- giá trị nằm trong khoảng: vd 1-5
 71            DECLARE
 72                range_parts INT[] = ARRAY[
 73                    CAST(SPLIT_PART(cron_parts[i], '-', 1) AS INT),
 74                    CAST(SPLIT_PART(cron_parts[i], '-', 2) AS INT)
 75                ];
 76            BEGIN
 77                IF target_value BETWEEN range_parts[1] AND range_parts[2] THEN
 78                    RETURN TRUE;
 79                END IF;
 80            END;
 81        ELSIF cron_parts[i] LIKE '*/%' THEN
 82            -- giá trị chia hết cho bước: vd */10
 83            DECLARE
 84                step_value INT = CAST(SPLIT_PART(cron_parts[i], '*/', 2) AS INT);
 85            BEGIN
 86                IF target_value % step_value = 0 THEN
 87                    RETURN TRUE;
 88                END IF;
 89            END;
 90        ELSIF cron_parts[i] LIKE '%|L' THEN
 91            -- ngày cụ thể, nếu tháng không có thì lấy ngày cuối tháng: vd 30|L
 92            DECLARE
 93                day_test INT = CAST(SPLIT_PART(cron_parts[i], '|', 1) AS INT);
 94                last_day_of_month INT = EXTRACT(DAY FROM DATE_TRUNC('MONTH', target_datetime + INTERVAL '1 MONTH') - INTERVAL '1 DAY');
 95            BEGIN
 96                IF target_value = day_test THEN
 97                    RETURN TRUE;
 98                END IF;
 99                IF day_test > last_day_of_month AND target_value = last_day_of_month THEN
100                    RETURN TRUE;
101                END IF;
102            END;
103        ELSIF cron_parts[i] = 'L' THEN
104            -- ngày cuối cùng của tháng
105            DECLARE
106                last_day_of_month INT = EXTRACT(DAY FROM DATE_TRUNC('MONTH', target_datetime + INTERVAL '1 MONTH') - INTERVAL '1 DAY');
107            BEGIN
108                IF target_value = last_day_of_month THEN
109                    RETURN TRUE;
110                END IF;
111            END;
112        ELSIF target_value = CAST(cron_parts[i] AS INT) THEN
113            -- đúng giá trị cụ thể
114            RETURN TRUE;
115        END IF;
116    END LOOP;
117
118    RETURN FALSE;
119END;
120$$ LANGUAGE plpgsql;

1SELECT td.is_cron_expression_satisfied(
2    '* * 6 30|L * * * 2026-01-01 2026-08-01',
3    NOW()
4);
5-- true nếu thời điểm hiện tại khớp lịch

 1DROP FUNCTION IF EXISTS td.get_job_to_run(
 2    target_datetime TIMESTAMP WITHOUT TIME ZONE,
 3    job_type VARCHAR
 4);
 5CREATE OR REPLACE FUNCTION td.get_job_to_run(
 6    target_datetime TIMESTAMP WITHOUT TIME ZONE,
 7    job_type VARCHAR
 8)
 9RETURNS SETOF td.job_user
10AS $$
11BEGIN
12    RETURN QUERY
13    SELECT *
14    FROM td.job_user j
15    WHERE j.job_type = job_type
16      AND j.is_active = TRUE
17      AND td.is_cron_expression_satisfied(j.cron_expression, target_datetime);
18END;
19$$ LANGUAGE plpgsql;

1SELECT *
2FROM td.get_job_to_run(
3    target_datetime => NOW(),       -- ngày giờ muốn check
4    job_type        => 'send_email' -- loại nghiệp vụ cần xử lý
5);

 1foreach (Job job in jobs)
 2{
 3    if (IsUserInactiveMoreThanDays(job, n: 7))
 4    {
 5        // không publish — đưa database id vào queue, gọi hàm dọn
 6        _concurrentQueueDBIds.Enqueue(job.DatabaseId);
 7        DoSlowMethod();
 8    }
 9    else
10    {
11        PublishToKafka(job);   // publish bình thường
12    }
13}

 1using System.Collections.Concurrent;
 2
 3namespace TDProject.Core.Business;
 4
 5/// <summary>
 6/// Chạy song song 2 luồng bằng kiểu dữ liệu ConcurrentQueue:
 7/// 1 thread thêm dữ liệu vào biến lưu trữ global,
 8/// 1 thread đọc dữ liệu ra và handle nghiệp vụ.
 9/// </summary>
10public class TwoThreadUsingConcurrentQueue
11{
12    /// <summary>
13    /// danh sách database id đang chờ xử lý
14    /// </summary>
15    private ConcurrentQueue<Guid> _concurrentQueueDBIds = new ConcurrentQueue<Guid>();
16
17    /// <summary>
18    /// cờ nhận biết có đang chạy lệnh ở thread xử lý nghiệp vụ không
19    /// </summary>
20    private bool _isRunningBussinessThread = false;
21
22    private void DoSlowMethod()
23    {
24        // nếu đã có task run rồi thì cứ chạy tiếp vòng while trong task đó
25        if (_isRunningBussinessThread)
26        {
27            return;
28        }
29
30        _isRunningBussinessThread = true;
31
32        // chưa có thì run task mới
33        Task.Run(() =>
34        {
35            try
36            {
37                while (_concurrentQueueDBIds.Count > 0)
38                {
39                    Guid currentDB;
40                    _concurrentQueueDBIds.TryDequeue(out currentDB);
41
42                    // luôn phải có try catch khi làm việc đa luồng
43                    try
44                    {
45                        // UPDATE td.job_user SET is_active = false WHERE database_id = @currentDB
46                        ExecuteAsync(
47                            "UPDATE td.job_user SET is_active = false WHERE database_id = @currentDB",
48                            currentDB);
49                    }
50                    catch (Exception ex)
51                    {
52                        Console.WriteLine($"{this.GetType()} {nameof(DoSlowMethod)}" + ex);
53                    }
54                }
55            }
56            catch (Exception ex)
57            {
58                Console.WriteLine($"{this.GetType()} {nameof(DoSlowMethod)}" + ex);
59            }
60            finally
61            {
62                _isRunningBussinessThread = false;
63            }
64        });
65    }
66}