use Activity types for user<->developer communication

Still all in a single process with no serialization, but now there are
separate threads for the user and developer sides, and they communicate
Activity back and forth.

Signatures are not checked yet, but both user and developer check that
the other side is sending Activity that forms a valid hash chain with
previous Activity.

The echo simulation is included, but may be buggy. This seems to work
well enough with 0 latency at least.

This commit was sponsored by Thom May on Patreon.

1 files changed, 180 insertions, 20 deletions

diff --git a/debug-me.hs b/debug-me.hs
index 69870f1..4122881 100644
--- a/debug-me.hs
+++ b/debug-me.hs
@@ -1,16 +1,19 @@
+{-# LANGUAGE OverloadedStrings #-}
+
 module Main where
 
 import Types
 import Hash
 import Pty
 
-import Control.Concurrent
 import Control.Concurrent.Async
 import Control.Concurrent.STM
 import System.IO
 import System.Process
 import System.Exit
 import qualified Data.ByteString as B
+import Data.List.NonEmpty (NonEmpty(..), toList)
+import Data.Monoid
 
 main :: IO ()
 main = do
@@ -22,25 +25,182 @@ main = do
 	go = withChildPty $ \p -> do
 		(Nothing, Nothing, Nothing, ph) <- createProcess $
 			onPty p (proc "bash" [])
-		let Master h = ptyMaster p
-		hSetBuffering stdin NoBuffering
-		hSetBuffering h NoBuffering
-		ithread <- async (forward stdin h ph 200000)
-		othread <- async (forward h stdout ph 0)
+		ichan <- newTChanIO
+		ochan <- newTChanIO
+		dthread <- async (developer ichan ochan)
+		uthread <- async (user p ichan ochan ph)
 		exitstatus <- waitForProcess ph
-		cancel ithread
-		cancel othread
+		cancel dthread
+		cancel uthread
 		return exitstatus
 
-forward :: Handle -> Handle -> ProcessHandle -> Int -> IO ()
-forward from to ph delay = do
-	b <- B.hGetSome from 1024
-	if b == B.empty
-		then do
-			terminateProcess ph
-			return ()
-		else do
-			threadDelay delay
-			B.hPut to b
-			hFlush to
-			forward from to ph delay
+developer :: TChan (Activity Entered) -> TChan (Activity Seen) -> IO ()
+developer ichan ochan = do
+	startact <- atomically $ readTChan ochan
+	case startact of
+		StartActivity (Seen b) sig -> do
+			B.hPut stdout b
+			hFlush stdout	
+		_ -> return ()
+	devstate <- newTVarIO (DeveloperState (hash startact) mempty)
+	_ <- sendTtyInput ichan devstate
+		`concurrently` sendTtyOutput ochan devstate
+	return ()
+
+data DeveloperState = DeveloperState
+	{ lastSeen :: HashPointer
+	, sentSince :: B.ByteString
+	}
+	deriving (Show)
+
+-- | Read things typed by the user, and forward them to the TChan.
+sendTtyInput :: TChan (Activity Entered) -> TVar DeveloperState -> IO ()
+sendTtyInput ichan devstate = go
+  where
+	go = do
+		b <- B.hGetSome stdin 1024
+		if b == B.empty
+			then return ()
+			else send b
+	send b = do
+		atomically $ do
+			ds <- readTVar devstate
+			let entered = Entered
+				{ enteredData = b
+				, echoData = sentSince ds
+				}
+			let act = Activity entered (lastSeen ds) dummySignature
+			writeTChan ichan act
+			let ds' = ds { sentSince = sentSince ds <> b }
+			writeTVar devstate ds'
+		go
+
+-- | Read activity from the TChan and display it.
+sendTtyOutput :: TChan (Activity Seen) -> TVar DeveloperState -> IO ()
+sendTtyOutput ochan devstate = go
+  where
+	go = do
+		b <- atomically get
+		B.hPut stdout b
+		hFlush stdout
+		go
+	get = do
+		act <- readTChan ochan
+		ds <- readTVar devstate
+		case act of
+			Activity (Seen b) hp sig
+				| hp == lastSeen ds -> do
+					let ds' = DeveloperState
+						{ lastSeen = hash act
+						-- XXX is this right?
+						-- What if we get
+						-- a Seen that is
+						-- part of what
+						-- sentSince
+						-- contained?
+						, sentSince = mempty
+						}
+					writeTVar devstate ds'
+					return b
+			-- Got an activity out of order or
+			-- missed one somehow. Ignore it?
+			_ -> return ("bad input")
+
+user :: Pty -> TChan (Activity Entered) -> TChan (Activity Seen) -> ProcessHandle -> IO ()
+user p ichan ochan ph = do
+	let startact = StartActivity (Seen ">>> debug-me started\r\n") dummySignature
+	atomically $ writeTChan ochan startact
+	backlog <- newTVarIO $ Backlog ((hash startact, startact) :| [])
+	_ <- sendPtyOutput p ochan backlog ph
+		`concurrently` sendPtyInput ichan p backlog
+	return ()
+
+-- | Log of recent output, with the most recent output first.
+data Backlog = Backlog (NonEmpty (HashPointer, Activity Seen))
+	deriving (Show)
+
+-- | Forward things written to the Pty out the TChan.
+sendPtyOutput :: Pty -> TChan (Activity Seen) -> TVar Backlog -> ProcessHandle -> IO ()
+sendPtyOutput p ochan backlog ph = go
+  where
+	Master h = ptyMaster p
+	go = do
+		b <- B.hGetSome h 1024
+		if b == B.empty
+			then do
+				terminateProcess ph
+				return ()
+			else send b
+	send b = do
+		atomically $ do
+			Backlog (bl@((prevhash, _) :| _)) <- readTVar backlog
+			let seen = Seen b
+			let act = Activity seen prevhash dummySignature
+			writeTChan ochan act
+			writeTVar backlog (Backlog ((hash act, act) :| toList bl))
+		go
+
+-- | Read things to be entered from the TChan, verify if they're legal,
+-- and send them to the Pty.
+sendPtyInput :: TChan (Activity Entered) -> Pty -> TVar Backlog -> IO ()
+sendPtyInput ichan p backlog = go
+  where
+	Master h = ptyMaster p
+	go = do
+		mb <- atomically $ do
+			newact <- readTChan ichan
+			bl <- readTVar backlog
+			-- Don't need to retain backlog before the Activity
+			-- that newact references.
+			let bl' = truncateBacklog bl newact
+			if isLegal newact bl'
+				then do
+					writeTVar backlog bl'
+					return (Right (enteredData (activityContent newact)))
+				else do
+					return (Left ("illegal entry", newact, bl'))
+		case mb of
+			Right b -> do
+				B.hPut h b
+				hFlush h
+				go
+			Left e -> do
+				print e
+				go
+
+-- | Truncate the Backlog to remove entries older than the one
+-- that the Activity refers to.
+--
+-- If the activity refers to an item not in the backlog, no truncation is
+-- done.
+truncateBacklog :: Backlog -> Activity Entered -> Backlog
+truncateBacklog (Backlog (b :| l)) (Activity _ hp _)
+	| fst b == hp = Backlog (b :| [])
+	| otherwise = Backlog (b :| go [] l)
+  where
+	go c [] = reverse c
+	go c (x:xs)
+		| fst x == hp = reverse (x:c)
+		| otherwise = go (x:c) xs
+truncateBacklog (Backlog bl) _ = Backlog bl
+
+-- | Entered activity is only legal if it points to the last Seen activvity,
+-- because this guarantees that the person who entered it saw
+-- the current state of the system before manipulating it.
+--
+-- To support typeahead on slow links, some echoData may be provided
+-- in the Entered activity. If the HashPointer in the activity points
+-- to an older Seen activity, then the echoData must match the
+-- concatenation of all activities after that one, up to the most recent
+-- Seen activity.
+isLegal :: Activity Entered -> Backlog -> Bool
+isLegal (Activity entered hp sig) (Backlog ((lastseenhash, _lastseen) :| bl))
+	| lastseenhash == hp = True
+	| any (== hp) (map fst bl) =
+		let sincehp = reverse (takeWhile (\(h, _) -> h /= hp) bl)
+		in echoData entered == mconcat (map (seenData . activityContent . snd) sincehp)
+	| otherwise = False
+isLegal _ _ = False
+
+dummySignature :: Signature
+dummySignature = Signature mempty