From d69953b05647e83c0c3a8211f06906ded074958f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Philip=20Dub=C3=A9?= <philip@peerdb.io>
Date: Sun, 5 May 2024 15:47:32 +0000
Subject: [PATCH] error dont panic, time

---
 flow/pua/peerdb.go         | 63 ++++++++++++++++++++++++++------------
 flow/pua/stream_adapter.go |  3 +-
 2 files changed, 45 insertions(+), 21 deletions(-)

diff --git a/flow/pua/peerdb.go b/flow/pua/peerdb.go
index 8e3221aba1..eb5df8f592 100644
--- a/flow/pua/peerdb.go
+++ b/flow/pua/peerdb.go
@@ -3,6 +3,7 @@ package pua
 import (
 	"bytes"
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/google/uuid"
@@ -137,6 +138,20 @@ func LuaRowIndex(ls *lua.LState) int {
 	return 1
 }
 
+func LVAsTime(ls *lua.LState, lv lua.LValue) time.Time {
+	switch v := lv.(type) {
+	case lua.LNumber:
+		ipart, fpart := math.Modf(float64(v))
+		return time.Unix(int64(ipart), int64(fpart*1e9))
+	case *lua.LUserData:
+		if tm, ok := v.Value.(time.Time); ok {
+			return tm
+		}
+	}
+	ls.RaiseError("Cannot convert %T to time.Time", lv)
+	return time.Time{}
+}
+
 func LuaRowNewIndex(ls *lua.LState) int {
 	_, row := LuaRow.Check(ls, 1)
 	key := ls.CheckString(2)
@@ -188,18 +203,16 @@ func LuaRowNewIndex(ls *lua.LState) int {
 		}
 	case qvalue.QValueKindString:
 		newqv = qvalue.QValueString{Val: lua.LVAsString(val)}
-		/* TODO time
-		case qvalue.QValueKindTimestamp:
-			newqv = qvalue.QValueTimestamp{Val:}
-		case qvalue.QValueKindTimestampTZ:
-			newqv = qvalue.QValueTimestampTZ{Val:}
-		case qvalue.QValueKindDate:
-			newqv = qvalue.QValueDate{Val:}
-		case qvalue.QValueKindTime:
-			newqv = qvalue.QValueTime{Val:}
-		case qvalue.QValueKindTimeTZ:
-			newqv = qvalue.QValueTimeTZ{Val:}
-		*/
+	case qvalue.QValueKindTimestamp:
+		newqv = qvalue.QValueTimestamp{Val: LVAsTime(ls, val)}
+	case qvalue.QValueKindTimestampTZ:
+		newqv = qvalue.QValueTimestampTZ{Val: LVAsTime(ls, val)}
+	case qvalue.QValueKindDate:
+		newqv = qvalue.QValueDate{Val: LVAsTime(ls, val)}
+	case qvalue.QValueKindTime:
+		newqv = qvalue.QValueTime{Val: LVAsTime(ls, val)}
+	case qvalue.QValueKindTimeTZ:
+		newqv = qvalue.QValueTimeTZ{Val: LVAsTime(ls, val)}
 	case qvalue.QValueKindNumeric:
 		if ud, ok := val.(*lua.LUserData); ok {
 			if num, ok := ud.Value.(decimal.Decimal); ok {
@@ -266,14 +279,24 @@ func LuaRowNewIndex(ls *lua.LState) int {
 				}),
 			}
 		}
-		/* TODO TIME
-		case qvalue.QValueKindArrayDate:
-			newqv = qvalue.QValueArrayDate{Val:}
-		case qvalue.QValueKindArrayTimestamp:
-			newqv = qvalue.QValueArrayTimestamp{Val:}
-		case qvalue.QValueKindArrayTimestampTZ:
-			newqv = qvalue.QValueArrayTimestampTZ{Val:}
-		*/
+	case qvalue.QValueKindArrayDate:
+		if tbl, ok := val.(*lua.LTable); ok {
+			newqv = qvalue.QValueArrayDate{
+				Val: shared.LTableToSlice(ls, tbl, LVAsTime),
+			}
+		}
+	case qvalue.QValueKindArrayTimestamp:
+		if tbl, ok := val.(*lua.LTable); ok {
+			newqv = qvalue.QValueArrayDate{
+				Val: shared.LTableToSlice(ls, tbl, LVAsTime),
+			}
+		}
+	case qvalue.QValueKindArrayTimestampTZ:
+		if tbl, ok := val.(*lua.LTable); ok {
+			newqv = qvalue.QValueArrayDate{
+				Val: shared.LTableToSlice(ls, tbl, LVAsTime),
+			}
+		}
 	case qvalue.QValueKindArrayBoolean:
 		if tbl, ok := val.(*lua.LTable); ok {
 			newqv = qvalue.QValueArrayBoolean{
diff --git a/flow/pua/stream_adapter.go b/flow/pua/stream_adapter.go
index 05e0371d14..0367134368 100644
--- a/flow/pua/stream_adapter.go
+++ b/flow/pua/stream_adapter.go
@@ -19,7 +19,8 @@ func AttachToStream(ls *lua.LState, lfn *lua.LFunction, stream *model.QRecordStr
 			ls.Push(lfn)
 			ls.Push(LuaRow.New(ls, row))
 			if err := ls.PCall(1, 0, nil); err != nil {
-				panic(err.Error()) // TODO error handling
+				output.Close(err)
+				return
 			}
 			for i, field := range schema.Fields {
 				record[i] = row.GetColumnValue(field.Name)