diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,30 @@
+Double Crown Gaming Co. (c) 2020
+
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+
+    * Redistributions in binary form must reproduce the above
+      copyright notice, this list of conditions and the following
+      disclaimer in the documentation and/or other materials provided
+      with the distribution.
+
+    * Neither the name of Double Crown Gaming Co. nor the names of other
+      contributors may be used to endorse or promote products derived
+      from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/benchmark/Main.hs b/benchmark/Main.hs
new file mode 100644
--- /dev/null
+++ b/benchmark/Main.hs
@@ -0,0 +1,22 @@
+module Main where
+
+import           Criterion.Main
+import           Data.Typeable  (Typeable)
+import           Dataflow
+import           Prelude
+import           Test.Dataflow  (runDataflow)
+
+
+main :: IO ()
+main = defaultMain [
+  bgroup "dataflow" [
+      bench "passthrough 1,000,000 inputs" $ nfIO (runDataflow passthrough [0..1000000 :: Int]),
+      bench "discard     1,000,000 inputs" $ nfIO (runDataflow blackhole [0..1000000 :: Int])
+    ]
+  ]
+  where
+    passthrough :: Typeable a => Edge a -> Dataflow (Edge a)
+    passthrough next = statelessVertex $ \t x -> send next t x
+
+    blackhole :: Typeable a => Edge a -> Dataflow (Edge a)
+    blackhole _ = statelessVertex $ \_ _ -> return ()
diff --git a/dataflower.cabal b/dataflower.cabal
new file mode 100644
--- /dev/null
+++ b/dataflower.cabal
@@ -0,0 +1,78 @@
+cabal-version: 1.12
+
+-- This file has been generated from package.yaml by hpack version 0.31.2.
+--
+-- see: https://github.com/sol/hpack
+--
+-- hash: 8f7b0ebc453194c8e6a5cca3c5cef03c62cfa135a28e2c15fcf3ff5a58e10d3a
+
+name:           dataflower
+version:        0.1.0.0
+synopsis:       A Pure-Haskell Timely Dataflow System
+description:    See README
+homepage:       https://github.com/doublecrowngaming/dataflower#readme
+bug-reports:    https://github.com/doublecrowngaming/dataflower/issues
+maintainer:     Jesse Kempf
+license:        BSD3
+license-file:   LICENSE
+build-type:     Simple
+
+source-repository head
+  type: git
+  location: https://github.com/doublecrowngaming/dataflower
+
+library
+  exposed-modules:
+      Dataflow
+      Test.Dataflow
+  other-modules:
+      Dataflow.Primitives
+      Dataflow.Vertices
+      Paths_dataflower
+  hs-source-dirs:
+      src
+  default-extensions: StrictData
+  ghc-options: -Wall -O2
+  build-depends:
+      base >=4.7 && <5
+    , hashable
+    , mtl
+    , stm
+    , time
+    , transformers
+    , vector
+  default-language: Haskell2010
+
+test-suite spec
+  type: exitcode-stdio-1.0
+  main-is: Spec.hs
+  other-modules:
+      DataflowSpec
+      Paths_dataflower
+  hs-source-dirs:
+      test
+  default-extensions: StrictData
+  ghc-options: -Wall -O2
+  build-depends:
+      QuickCheck
+    , base >=4.7 && <5
+    , dataflower
+    , hspec >=2.0.0
+    , stm
+  default-language: Haskell2010
+
+benchmark performance
+  type: exitcode-stdio-1.0
+  main-is: Main.hs
+  other-modules:
+      Paths_dataflower
+  hs-source-dirs:
+      benchmark
+  default-extensions: StrictData
+  ghc-options: -Wall -O2 -O2
+  build-depends:
+      base >=4.7 && <5
+    , criterion
+    , dataflower
+    , stm
+  default-language: Haskell2010
diff --git a/src/Dataflow.hs b/src/Dataflow.hs
new file mode 100644
--- /dev/null
+++ b/src/Dataflow.hs
@@ -0,0 +1,55 @@
+{-# LANGUAGE RecordWildCards #-}
+
+{-|
+Module      : Dataflow
+Description : Timely Dataflow for Haskell
+Copyright   : (c) Double Crown Gaming Co. 2020
+License     : BSD3
+Maintainer  : jesse.kempf@doublecrown.co
+Stability   : experimental
+
+Timely Dataflow in pure Haskell.
+-}
+
+module Dataflow (
+  Dataflow,
+  Edge,
+  Timestamp,
+  StateRef,
+  send,
+  readState,
+  writeState,
+  modifyState,
+  statefulVertex,
+  statelessVertex,
+  outputTVar,
+  Program,
+  compile,
+  execute
+) where
+
+import           Control.Monad              (void)
+import           Control.Monad.State.Strict (execStateT, runStateT)
+import           Data.Traversable           (Traversable)
+import           Data.Typeable              (Typeable)
+import           Dataflow.Primitives
+import           Dataflow.Vertices
+
+-- | A 'Program' represents a fully-preprocessed 'Dataflow' that may be
+-- executed against inputs.
+data Program i = Program {
+  programInput :: Edge i,
+  programState :: DataflowState
+}
+
+-- | Take a 'Dataflow' which takes 'i's as input and compile it into a 'Program'.
+compile :: Dataflow (Edge i) -> IO (Program i)
+compile (Dataflow actions) = uncurry Program <$> runStateT actions initDataflowState
+
+-- | Feed a traversable collection of inputs to a 'Program'. All inputs provided will
+-- have the same 'Timestamp' associated with them.
+execute :: (Traversable t, Typeable i) => t i -> Program i -> IO ()
+execute corpus Program{..} = execDataflow feedInput
+  where
+    feedInput           = input corpus programInput
+    execDataflow action = void $ execStateT (runDataflow action) programState
diff --git a/src/Dataflow/Primitives.hs b/src/Dataflow/Primitives.hs
new file mode 100644
--- /dev/null
+++ b/src/Dataflow/Primitives.hs
@@ -0,0 +1,166 @@
+{-# LANGUAGE ExistentialQuantification  #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE MultiParamTypeClasses      #-}
+
+module Dataflow.Primitives (
+  Dataflow(..),
+  DataflowState,
+  Vertex(..),
+  initDataflowState,
+  StateRef,
+  newState,
+  readState,
+  writeState,
+  modifyState,
+  Edge,
+  Timestamp(..),
+  registerVertex,
+  registerFinalizer,
+  incrementEpoch,
+  input,
+  send,
+  finalize
+) where
+
+import           Control.Arrow              ((>>>))
+import           Control.Monad.State.Strict (StateT, gets, modify)
+import           Control.Monad.Trans        (lift)
+import           Data.Dynamic               (Dynamic, Typeable, fromDyn, toDyn)
+import           Data.Hashable              (Hashable (..))
+import           Data.IORef                 (IORef, atomicModifyIORef',
+                                             atomicWriteIORef, newIORef,
+                                             readIORef)
+import           Data.Vector                (Vector, empty, snoc, (!))
+import           Numeric.Natural            (Natural)
+import           Prelude
+
+
+newtype VertexID    = VertexID        Int deriving (Eq, Ord, Show)
+newtype Epoch       = Epoch       Natural deriving (Eq, Ord, Hashable, Show)
+
+-- | 'Timestamp's represent instants in the causal timeline.
+newtype Timestamp   = Timestamp     Epoch deriving (Eq, Ord, Hashable, Show)
+
+-- | An 'Edge' is a typed reference to a computational vertex that
+-- takes 'a's as its input.
+newtype Edge a      = Edge       VertexID
+
+-- | Class of entities that can be incremented by one.
+class Incrementable a where
+  inc :: a -> a
+
+instance Incrementable VertexID where
+  inc (VertexID n) = VertexID (n + 1)
+
+instance Incrementable Epoch where
+  inc (Epoch n) = Epoch (n + 1)
+
+
+data DataflowState = DataflowState {
+  dfsVertices       :: Vector Dynamic,
+  dfsFinalizers     :: [Timestamp -> Dataflow ()],
+  dfsLastVertexID   :: VertexID,
+  dfsLastInputEpoch :: Epoch
+}
+
+-- | `Dataflow` is the type of all dataflow operations.
+newtype Dataflow a = Dataflow { runDataflow :: StateT DataflowState IO a }
+  deriving (Functor, Applicative, Monad)
+
+initDataflowState :: DataflowState
+initDataflowState = DataflowState {
+  dfsVertices       = empty,
+  dfsFinalizers     = [],
+  dfsLastVertexID   = VertexID (-1),
+  dfsLastInputEpoch = Epoch 0
+}
+
+-- | Get the next input Epoch.
+incrementEpoch :: Dataflow Epoch
+incrementEpoch =
+  Dataflow $ do
+    epoch <- gets (dfsLastInputEpoch >>> inc)
+
+    modify $ \s -> s { dfsLastInputEpoch = epoch }
+
+    return epoch
+
+
+data Vertex i = forall s.
+    StatefulVertex
+      (StateRef s)
+      (StateRef s -> Timestamp -> i -> Dataflow ())
+  | StatelessVertex
+      (Timestamp -> i -> Dataflow ())
+
+-- | Retrieve the vertex for a given edge.
+lookupVertex :: Typeable i => Edge i -> Dataflow (Vertex i)
+lookupVertex (Edge (VertexID vindex)) =
+  Dataflow $ do
+    vertices <- gets dfsVertices
+
+    return $ fromDyn (vertices ! vindex) (error "Programming error: Vertex and Edge were of different types")
+
+-- | Store a provided vertex and obtain an 'Edge' that refers to it.
+registerVertex :: Typeable i => Vertex i -> Dataflow (Edge i)
+registerVertex vertex =
+  Dataflow $ do
+    vid <- gets (dfsLastVertexID >>> inc)
+
+    modify $ addVertex vertex vid
+
+    return (Edge vid)
+
+  where
+    addVertex vtx vid s = s {
+      dfsVertices     = dfsVertices s `snoc` toDyn vtx,
+      dfsLastVertexID = vid
+    }
+
+-- | Store a provided finalizer.
+registerFinalizer :: (Timestamp -> Dataflow ()) -> Dataflow ()
+registerFinalizer finalizer =
+  Dataflow $ modify $ \s -> s { dfsFinalizers = finalizer : dfsFinalizers s }
+
+-- | Mutable state that holds an `a`.
+newtype StateRef a = StateRef (IORef a)
+
+-- | Create a `StateRef` initialized to the provided `a`.
+newState :: a -> Dataflow (StateRef a)
+newState a = StateRef <$> (Dataflow $ lift $ newIORef a)
+
+-- | Read the value stored in the `StateRef`.
+readState :: StateRef a -> Dataflow a
+readState (StateRef ref) = Dataflow $ lift (readIORef ref)
+
+-- | Overwrite the value stored in the `StateRef`.
+writeState :: StateRef a -> a -> Dataflow ()
+writeState (StateRef ref) x = Dataflow $ lift $ atomicWriteIORef ref x
+
+-- | Update the value stored in `StateRef`.
+modifyState :: StateRef a -> (a -> a) -> Dataflow ()
+modifyState (StateRef ref) op = Dataflow $ lift $ atomicModifyIORef' ref (\x -> (op x, ()))
+
+{-# INLINEABLE input #-}
+input :: (Traversable t, Typeable i) => t i -> Edge i -> Dataflow ()
+input inputs next = do
+  timestamp <- Timestamp <$> incrementEpoch
+
+  mapM_ (send next timestamp) inputs
+
+  finalize timestamp
+
+{-# NOINLINE send #-}
+-- | Send an `input` item to be worked on to the indicated vertex.
+send :: Typeable input => Edge input -> Timestamp -> input -> Dataflow ()
+send e t i = lookupVertex e >>= invoke t i
+  where
+    invoke timestamp datum (StatefulVertex sref callback) = callback sref timestamp datum
+    invoke timestamp datum (StatelessVertex callback)     = callback timestamp datum
+
+-- Notify all relevant vertices that no more input is coming for `Timestamp`.
+finalize :: Timestamp -> Dataflow ()
+finalize t = do
+  finalizers <- Dataflow $ gets dfsFinalizers
+
+  mapM_ (\p -> p t) finalizers
diff --git a/src/Dataflow/Vertices.hs b/src/Dataflow/Vertices.hs
new file mode 100644
--- /dev/null
+++ b/src/Dataflow/Vertices.hs
@@ -0,0 +1,52 @@
+{-# LANGUAGE ExplicitForAll      #-}
+{-# LANGUAGE OverloadedLists     #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Dataflow.Vertices (
+  statefulVertex,
+  statelessVertex,
+  outputTVar
+) where
+
+import           Control.Concurrent.STM.TVar (TVar, modifyTVar')
+import           Control.Monad.STM           (atomically)
+import           Control.Monad.Trans.Class   (lift)
+import           Data.Typeable               (Typeable)
+import           Dataflow.Primitives         (Dataflow (..), Edge, StateRef,
+                                              Timestamp (..), Vertex (..),
+                                              newState, registerFinalizer,
+                                              registerVertex)
+import           Prelude
+
+
+-- | Construct a vertex with internal state. Like 'statelessVertex', 'statefulVertex'
+-- requires a procedure to invoke on each input. It also needs an initial 'state' value
+-- and a procedure to call when all inputs for a given 'Timestamp' value have been
+-- delivered.
+--
+-- NB: Until the finalizer has been called for a particular timestamp, a stateful vertex
+-- must be capable of accepting data for multiple timestamps simultaneously.
+statefulVertex :: Typeable i =>
+  state -- ^ The initial state value.
+  -> (StateRef state -> Timestamp -> i -> Dataflow ()) -- ^ The input handler.
+  -> (StateRef state -> Timestamp -> Dataflow ()) -- ^ The finalizer.
+  -> Dataflow (Edge i)
+statefulVertex initState callback finalizer = do
+  stateRef <- newState initState
+
+  registerFinalizer $ finalizer stateRef
+  registerVertex    $ StatefulVertex stateRef callback
+
+-- | Construct a vertex with no internal state. The given procedure is invoked on each input.
+--
+-- `send`ing to a stateless vertex is effectively a function call and will execute in the
+-- caller's thread. By design this is a cheap operation.
+statelessVertex :: Typeable i => (Timestamp -> i -> Dataflow ()) -> Dataflow (Edge i)
+statelessVertex callback = registerVertex $ StatelessVertex callback
+
+{-# NOINLINE outputTVar #-}
+-- | Construct an output vertex that stores items into the provided 'TVar'. The first argument
+-- is an update function so that, for example, the 'TVar' could contain a list of 'o's and the update
+-- function could then `cons` new items onto the list.
+outputTVar :: Typeable o => (o -> w -> w) -> TVar w -> Dataflow (Edge o)
+outputTVar op register = statelessVertex $ \_ x -> Dataflow $ lift $ atomically $ modifyTVar' register (op x)
diff --git a/src/Test/Dataflow.hs b/src/Test/Dataflow.hs
new file mode 100644
--- /dev/null
+++ b/src/Test/Dataflow.hs
@@ -0,0 +1,21 @@
+module Test.Dataflow (
+  runDataflow
+) where
+
+import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO)
+import           Control.Monad.IO.Class      (MonadIO (..))
+import           Data.Typeable               (Typeable)
+import           Dataflow                    (Dataflow, Edge, compile, execute,
+                                              outputTVar)
+import           Prelude
+
+
+runDataflow :: (Typeable i, Typeable o, MonadIO io) => (Edge o -> Dataflow (Edge i)) -> [i] -> io [o]
+runDataflow dataflow inputs =
+  liftIO $ do
+    out     <- newTVarIO []
+    program <- compile (dataflow =<< outputTVar (:) out)
+
+    execute inputs program
+
+    reverse <$> readTVarIO out
diff --git a/test/DataflowSpec.hs b/test/DataflowSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/DataflowSpec.hs
@@ -0,0 +1,36 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module DataflowSpec (spec) where
+
+import           Control.Monad   ((>=>))
+import           Data.Typeable   (Typeable)
+import           Dataflow
+import           Prelude         hiding (map)
+
+import           Test.Dataflow   (runDataflow)
+import           Test.Hspec
+import           Test.QuickCheck
+
+
+spec :: Spec
+spec = do
+  it "can pass through data without modification" $ property $ do
+    let passthrough next = statelessVertex $ \t x -> send next t x
+
+    \(numbers :: [Integer]) -> runDataflow passthrough numbers `shouldReturn` numbers
+
+  describe "finalize" $ do
+    it "finalizes vertices" $ property $
+      \(numbers :: [Int]) -> runDataflow storeAndForward numbers `shouldReturn` numbers
+
+    it "finalizes vertices in the correct order" $ property $
+      \(numbers :: [Int]) ->
+        runDataflow (storeAndForward >=> storeAndForward >=> storeAndForward) numbers `shouldReturn` numbers
+
+storeAndForward :: (Show i, Typeable i) => Edge i -> Dataflow (Edge i)
+storeAndForward next = statefulVertex [] store forward
+  where
+    store sref _ i = modifyState sref (i :)
+    forward sref t = do
+      mapM_ (send next t) =<< reverse <$> readState sref
+      writeState sref []
diff --git a/test/Spec.hs b/test/Spec.hs
new file mode 100644
--- /dev/null
+++ b/test/Spec.hs
@@ -0,0 +1,1 @@
+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}