From 96c07bab1086510a55a004e9a1b90c30bc723a5c Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Sat, 29 Aug 2026 01:08:28 +0700
Subject: feat: init base draft support window

---
 core/record.go | 40 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 40 insertions(+)
 create mode 100644 core/record.go

(limited to 'core/record.go')

diff --git a/core/record.go b/core/record.go
new file mode 100644
index 0000000..e18e58d
--- /dev/null
+++ b/core/record.go
@@ -0,0 +1,40 @@
+package core
+
+type RecordType byte
+
+const (
+	RecordData RecordType = iota
+	RecordWatermark
+	RecordBarrier
+)
+
+type StreamRecord[T any] struct {
+	Value     T
+	Key       string // for keyBy stateful processing
+	Timestamp int64  // epoch ms time
+	BarrierID uint64 // ID checkpoint barrier
+	Type      RecordType
+}
+
+func NewDataRecord[T any](key string, val T, timestamp int64) StreamRecord[T] {
+	return StreamRecord[T]{
+		Value:     val,
+		Key:       key,
+		Timestamp: timestamp,
+		Type:      RecordData,
+	}
+}
+
+func NewWatermarkRecord[T any](timestamp int64) StreamRecord[T] {
+	return StreamRecord[T]{
+		Type:      RecordWatermark,
+		Timestamp: timestamp,
+	}
+}
+
+func NewBarrierRecord[T any](barrierID uint64) StreamRecord[T] {
+	return StreamRecord[T]{
+		Type:      RecordBarrier,
+		BarrierID: barrierID,
+	}
+}
-- 
cgit v1.3.1