fuck off

34 files changed, 843 insertions, 537 deletions

diff --git a/src/app_console/app_console.cpp b/src/app_console/app_console.cpp
index a3f04bf9..74e11de5 100644
--- a/src/app_console/app_console.cpp
+++ b/src/app_console/app_console.cpp
@@ -21,6 +21,7 @@
 #include "audio_fsm.hpp"
 #include "database.hpp"
 #include "esp_console.h"
+#include "esp_intr_alloc.h"
 #include "esp_log.h"
 #include "event_queue.hpp"
 #include "ff.h"
diff --git a/src/audio/CMakeLists.txt b/src/audio/CMakeLists.txt
index 2501f773..6ef144ac 100644
--- a/src/audio/CMakeLists.txt
+++ b/src/audio/CMakeLists.txt
@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: GPL-3.0-only
 
 idf_component_register(
-  SRCS "audio_decoder.cpp" "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.cpp"
+  SRCS "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.cpp"
   "stream_message.cpp" "i2s_audio_output.cpp" "stream_buffer.cpp" "track_queue.cpp"
   "stream_event.cpp" "pipeline.cpp" "stream_info.cpp" "audio_fsm.cpp"
   INCLUDE_DIRS "include"
diff --git a/src/audio/audio_fsm.cpp b/src/audio/audio_fsm.cpp
index 5f4f8783..c3313820 100644
--- a/src/audio/audio_fsm.cpp
+++ b/src/audio/audio_fsm.cpp
@@ -14,6 +14,8 @@
 #include "esp_log.h"
 #include "event_queue.hpp"
 #include "fatfs_audio_input.hpp"
+#include "freertos/portmacro.h"
+#include "future_fetcher.hpp"
 #include "i2s_audio_output.hpp"
 #include "i2s_dac.hpp"
 #include "pipeline.hpp"
@@ -29,14 +31,16 @@ drivers::IGpios* AudioState::sIGpios;
 std::shared_ptr<drivers::I2SDac> AudioState::sDac;
 std::weak_ptr<database::Database> AudioState::sDatabase;
 
+std::unique_ptr<AudioTask> AudioState::sTask;
 std::unique_ptr<FatfsAudioInput> AudioState::sFileSource;
 std::unique_ptr<I2SAudioOutput> AudioState::sI2SOutput;
-std::vector<std::unique_ptr<IAudioElement>> AudioState::sPipeline;
 
 TrackQueue* AudioState::sTrackQueue;
+std::optional<database::TrackId> AudioState::sCurrentTrack;
 
 auto AudioState::Init(drivers::IGpios* gpio_expander,
                       std::weak_ptr<database::Database> database,
+                      std::shared_ptr<database::ITagParser> tag_parser,
                       TrackQueue* queue) -> bool {
   sIGpios = gpio_expander;
   sTrackQueue = queue;
@@ -48,19 +52,10 @@ auto AudioState::Init(drivers::IGpios* gpio_expander,
   sDac.reset(dac.value());
   sDatabase = database;
 
-  sFileSource.reset(new FatfsAudioInput());
+  sFileSource.reset(new FatfsAudioInput(tag_parser));
   sI2SOutput.reset(new I2SAudioOutput(sIGpios, sDac));
 
-  // Perform initial pipeline configuration.
-  // TODO(jacqueline): Factor this out once we have any kind of dynamic
-  // reconfiguration.
-  AudioDecoder* codec = new AudioDecoder();
-  sPipeline.emplace_back(codec);
-
-  Pipeline* pipeline = new Pipeline(sPipeline.front().get());
-  pipeline->AddInput(sFileSource.get());
-
-  task::StartPipeline(pipeline, sI2SOutput.get());
+  AudioTask::Start(sFileSource.get(), sI2SOutput.get());
 
   return true;
 }
@@ -85,9 +80,9 @@ void AudioState::react(const system_fsm::KeyDownChanged& ev) {
 
 void AudioState::react(const system_fsm::HasPhonesChanged& ev) {
   if (ev.falling) {
-    ESP_LOGI(kTag, "headphones in!");
+    // ESP_LOGI(kTag, "headphones in!");
   } else {
-    ESP_LOGI(kTag, "headphones out!");
+    // ESP_LOGI(kTag, "headphones out!");
   }
 }
 
@@ -107,13 +102,15 @@ void Standby::react(const QueueUpdate& ev) {
     return;
   }
 
+  sCurrentTrack = current_track;
+
   auto db = sDatabase.lock();
   if (!db) {
     ESP_LOGW(kTag, "database not open; ignoring play request");
     return;
   }
 
-  sFileSource->OpenFile(db->GetTrackPath(*current_track));
+  sFileSource->SetPath(db->GetTrackPath(*current_track));
 }
 
 void Playback::entry() {
@@ -127,20 +124,25 @@ void Playback::exit() {
 }
 
 void Playback::react(const QueueUpdate& ev) {
+  if (!ev.current_changed) {
+    return;
+  }
   auto current_track = sTrackQueue->GetCurrent();
   if (!current_track) {
-    // TODO: return to standby?
+    sFileSource->SetPath();
+    sCurrentTrack.reset();
+    transit<Standby>();
     return;
   }
 
+  sCurrentTrack = current_track;
+
   auto db = sDatabase.lock();
   if (!db) {
     return;
   }
 
-  // TODO: what if we just finished this, and are preemptively loading the next
-  // one?
-  sFileSource->OpenFile(db->GetTrackPath(*current_track));
+  sFileSource->SetPath(db->GetTrackPath(*current_track));
 }
 
 void Playback::react(const PlaybackUpdate& ev) {
@@ -161,7 +163,7 @@ void Playback::react(const internal::InputFileClosed& ev) {
     return;
   }
   ESP_LOGI(kTag, "preemptively opening next file");
-  sFileSource->OpenFile(db->GetTrackPath(upcoming.front()));
+  sFileSource->SetPath(db->GetTrackPath(upcoming.front()));
 }
 
 void Playback::react(const internal::InputFileFinished& ev) {
diff --git a/src/audio/audio_task.cpp b/src/audio/audio_task.cpp
index babe6849..dbe5d50e 100644
--- a/src/audio/audio_task.cpp
+++ b/src/audio/audio_task.cpp
@@ -9,23 +9,29 @@
 #include <stdlib.h>
 
 #include <algorithm>
+#include <cmath>
 #include <cstddef>
 #include <cstdint>
+#include <cstring>
 #include <deque>
 #include <memory>
 #include <variant>
 
+#include "audio_decoder.hpp"
 #include "audio_events.hpp"
 #include "audio_fsm.hpp"
 #include "audio_sink.hpp"
 #include "cbor.h"
+#include "codec.hpp"
 #include "esp_err.h"
 #include "esp_heap_caps.h"
 #include "esp_log.h"
 #include "event_queue.hpp"
+#include "fatfs_audio_input.hpp"
 #include "freertos/portmacro.h"
 #include "freertos/projdefs.h"
 #include "freertos/queue.h"
+#include "freertos/ringbuf.h"
 #include "pipeline.hpp"
 #include "span.hpp"
 
@@ -41,193 +47,209 @@
 
 namespace audio {
 
-namespace task {
-
-static const char* kTag = "task";
-
-// The default amount of time to wait between pipeline iterations for a single
-// track.
-static constexpr uint_fast16_t kDefaultDelayTicks = pdMS_TO_TICKS(5);
-static constexpr uint_fast16_t kMaxDelayTicks = pdMS_TO_TICKS(10);
-static constexpr uint_fast16_t kMinDelayTicks = pdMS_TO_TICKS(1);
-
-void AudioTaskMain(std::unique_ptr<Pipeline> pipeline, IAudioSink* sink) {
-  // The stream format for bytes currently in the sink buffer.
-  std::optional<StreamInfo::Format> output_format;
-
-  // How long to wait between pipeline iterations. This is reset for each track,
-  // and readjusted on the fly to maintain a reasonable amount playback buffer.
-  // Buffering too much will mean we process samples inefficiently, wasting CPU
-  // time, whilst buffering too little will affect the quality of the output.
-  uint_fast16_t delay_ticks = kDefaultDelayTicks;
-
-  std::vector<Pipeline*> all_elements = pipeline->GetIterationOrder();
-
-  float current_sample_in_second = 0;
-  uint32_t previous_second = 0;
-  uint32_t current_second = 0;
-
-  bool previously_had_work = false;
-  events::EventQueue& event_queue = events::EventQueue::GetInstance();
-  while (1) {
-    // First, see if we actually have any pipeline work to do in this iteration.
-    bool has_work = false;
-    // We always have work to do if there's still bytes to be sunk.
-    has_work = all_elements.back()->OutStream().info->bytes_in_stream > 0;
-    if (!has_work) {
-      for (Pipeline* p : all_elements) {
-        has_work = p->OutputElement()->NeedsToProcess();
-        if (has_work) {
-          break;
-        }
-      }
-    }
-
-    if (!has_work) {
-      has_work = !xStreamBufferIsEmpty(sink->buffer());
-    }
-
-    if (previously_had_work && !has_work) {
-      events::Dispatch<internal::AudioPipelineIdle, AudioState>({});
-    }
-    previously_had_work = has_work;
-
-    // See if there's any new events.
-    event_queue.ServiceAudio(has_work ? delay_ticks : portMAX_DELAY);
-
-    if (!has_work) {
-      // See if we've been given work by this event.
-      for (Pipeline* p : all_elements) {
-        has_work = p->OutputElement()->NeedsToProcess();
-        if (has_work) {
-          delay_ticks = kDefaultDelayTicks;
-          break;
-        }
-      }
-      if (!has_work) {
-        continue;
-      }
-    }
-
-    // We have work to do! Allow each element in the pipeline to process one
-    // chunk. We iterate from input nodes first, so this should result in
-    // samples in the output buffer.
-
-    for (int i = 0; i < all_elements.size(); i++) {
-      std::vector<RawStream> raw_in_streams;
-      all_elements.at(i)->InStreams(&raw_in_streams);
-      RawStream raw_out_stream = all_elements.at(i)->OutStream();
-
-      // Crop the input and output streams to the ranges that are safe to
-      // touch. For the input streams, this is the region that contains
-      // data. For the output stream, this is the region that does *not*
-      // already contain data.
-      std::vector<InputStream> in_streams;
-      std::for_each(raw_in_streams.begin(), raw_in_streams.end(),
-                    [&](RawStream& s) { in_streams.emplace_back(&s); });
-      OutputStream out_stream(&raw_out_stream);
-
-      all_elements.at(i)->OutputElement()->Process(in_streams, &out_stream);
-    }
-
-    RawStream raw_sink_stream = all_elements.back()->OutStream();
-    InputStream sink_stream(&raw_sink_stream);
-
-    if (sink_stream.info().bytes_in_stream == 0) {
-      if (sink_stream.is_producer_finished()) {
-        sink_stream.mark_consumer_finished();
-
-        if (current_second > 0 || current_sample_in_second > 0) {
-          events::Dispatch<internal::InputFileFinished, AudioState>({});
-        }
-
-        current_second = 0;
-        previous_second = 0;
-        current_sample_in_second = 0;
-      } else {
-        // The user is probably about to hear a skip :(
-        ESP_LOGW(kTag, "!! audio sink is underbuffered !!");
-      }
-      // No new bytes to sink, so skip sinking completely.
-      continue;
-    }
-
-    if (!output_format || output_format != sink_stream.info().format) {
-      // The format of the stream within the sink stream has changed. We
-      // need to reconfigure the sink, but shouldn't do so until we've fully
-      // drained the current buffer.
-      if (xStreamBufferIsEmpty(sink->buffer())) {
-        ESP_LOGI(kTag, "reconfiguring dac");
-        output_format = sink_stream.info().format;
-        sink->Configure(*output_format);
-      } else {
-        ESP_LOGI(kTag, "waiting to reconfigure");
-        continue;
-      }
-    }
-
-    // We've reconfigured the sink, or it was already configured correctly.
-    // Send through some data.
-    std::size_t bytes_sunk =
-        xStreamBufferSend(sink->buffer(), sink_stream.data().data(),
-                          sink_stream.data().size_bytes(), 0);
-
-    if (std::holds_alternative<StreamInfo::Pcm>(*output_format)) {
-      StreamInfo::Pcm pcm = std::get<StreamInfo::Pcm>(*output_format);
-
-      float samples_sunk = bytes_sunk;
-      samples_sunk /= pcm.channels;
-
-      // Samples must be aligned to 16 bits. The number of actual bytes per
-      // sample is therefore the bps divided by 16, rounded up (align to word),
-      // times two (convert to bytes).
-      uint8_t bytes_per_sample = ((pcm.bits_per_sample + 16 - 1) / 16) * 2;
-      samples_sunk /= bytes_per_sample;
-
-      current_sample_in_second += samples_sunk;
-      while (current_sample_in_second >= pcm.sample_rate) {
-        current_second++;
-        current_sample_in_second -= pcm.sample_rate;
-      }
-      if (previous_second != current_second) {
-        events::Dispatch<PlaybackUpdate, AudioState, ui::UiState>({
-            .seconds_elapsed = current_second,
-            .seconds_total =
-                sink_stream.info().duration_seconds.value_or(current_second),
-        });
-      }
-      previous_second = current_second;
-    }
-
-    // Adjust how long we wait for the next iteration if we're getting too far
-    // ahead or behind.
-    float sunk_percent = static_cast<float>(bytes_sunk) /
-                         static_cast<float>(sink_stream.info().bytes_in_stream);
-
-    if (sunk_percent > 0.66f) {
-      // We're sinking a lot of the output buffer per iteration, so we need to
-      // be running faster.
-      delay_ticks--;
-    } else if (sunk_percent < 0.33f) {
-      // We're not sinking much of the output buffer per iteration, so we can
-      // slow down to save some cycles.
-      delay_ticks++;
-    }
-    delay_ticks = std::clamp(delay_ticks, kMinDelayTicks, kMaxDelayTicks);
-
-    // Finally, actually mark the bytes we sunk as consumed.
-    if (bytes_sunk > 0) {
-      sink_stream.consume(bytes_sunk);
-    }
+static const char* kTag = "audio_dec";
+
+static constexpr std::size_t kSampleBufferSize = 16 * 1024;
+
+Timer::Timer(StreamInfo::Pcm format)
+    : format_(format),
+      last_seconds_(0),
+      total_duration_seconds_(0),
+      current_seconds_(0) {}
+
+auto Timer::SetLengthSeconds(uint32_t len) -> void {
+  total_duration_seconds_ = len;
+}
+
+auto Timer::SetLengthBytes(uint32_t len) -> void {
+  total_duration_seconds_ = 0;
+}
+
+auto Timer::AddBytes(std::size_t bytes) -> void {
+  float samples_sunk = bytes;
+  samples_sunk /= format_.channels;
+
+  // Samples must be aligned to 16 bits. The number of actual bytes per
+  // sample is therefore the bps divided by 16, rounded up (align to word),
+  // times two (convert to bytes).
+  uint8_t bytes_per_sample = ((format_.bits_per_sample + 16 - 1) / 16) * 2;
+  samples_sunk /= bytes_per_sample;
+
+  current_seconds_ += samples_sunk / format_.sample_rate;
+
+  uint32_t rounded = std::round(current_seconds_);
+  if (rounded != last_seconds_) {
+    last_seconds_ = rounded;
+    events::Dispatch<PlaybackUpdate, AudioState, ui::UiState>(PlaybackUpdate{
+        .seconds_elapsed = rounded,
+        .seconds_total =
+            total_duration_seconds_ == 0 ? rounded : total_duration_seconds_});
   }
 }
 
-auto StartPipeline(Pipeline* pipeline, IAudioSink* sink) -> void {
-  ESP_LOGI(kTag, "starting audio pipeline task");
-  tasks::StartPersistent<tasks::Type::kAudio>(
-      [=]() { AudioTaskMain(std::unique_ptr<Pipeline>(pipeline), sink); });
+auto AudioTask::Start(IAudioSource* source, IAudioSink* sink) -> AudioTask* {
+  AudioTask* task = new AudioTask(source, sink);
+  tasks::StartPersistent<tasks::Type::kAudio>([=]() { task->Main(); });
+  return task;
 }
 
-}  // namespace task
+AudioTask::AudioTask(IAudioSource* source, IAudioSink* sink)
+    : source_(source),
+      sink_(sink),
+      codec_(),
+      timer_(),
+      is_new_stream_(false),
+      current_input_format_(),
+      current_output_format_(),
+      sample_buffer_(reinterpret_cast<std::byte*>(
+          heap_caps_malloc(kSampleBufferSize,
+                           MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT))),
+      sample_buffer_len_(kSampleBufferSize) {}
+
+void AudioTask::Main() {
+  for (;;) {
+    source_->Read(
+        [this](StreamInfo::Format format) -> bool {
+          if (current_input_format_ && format == *current_input_format_) {
+            // This is the continuation of previous data. We can handle it if
+            // we are able to decode it, or if it doesn't need decoding.
+            return current_output_format_ == format || codec_ != nullptr;
+          }
+          // This must be a new stream of data. Reset everything to prepare to
+          // handle it.
+          current_input_format_ = format;
+          is_new_stream_ = true;
+          codec_.reset();
+          timer_.reset();
+
+          // What kind of data does this new stream contain?
+          if (std::holds_alternative<StreamInfo::Pcm>(format)) {
+            // It's already decoded! We can handle this immediately if it
+            // matches what we're currently sending to the sink. Otherwise, we
+            // will need to wait for the sink to drain before we can reconfigure
+            // it.
+            if (current_output_format_ && format == *current_output_format_) {
+              return true;
+            } else if (xStreamBufferIsEmpty(sink_->stream())) {
+              return true;
+            } else {
+              return false;
+            }
+          } else if (std::holds_alternative<StreamInfo::Encoded>(format)) {
+            // The stream has some kind of encoding. Whether or not we can
+            // handle it is entirely down to whether or not we have a codec for
+            // it.
+            auto encoding = std::get<StreamInfo::Encoded>(format);
+            auto codec = codecs::CreateCodecForType(encoding.type);
+            if (codec) {
+              ESP_LOGI(kTag, "successfully created codec for stream");
+              codec_.reset(*codec);
+              return true;
+            } else {
+              ESP_LOGE(kTag, "stream has unknown encoding");
+              return false;
+            }
+          } else {
+            // programmer error / skill issue :(
+            ESP_LOGE(kTag, "stream has unknown format");
+            current_input_format_ = format;
+            return false;
+          }
+        },
+        [this](cpp::span<const std::byte> bytes) -> size_t {
+          // PCM streams are simple, so handle them first.
+          if (std::holds_alternative<StreamInfo::Pcm>(*current_input_format_)) {
+            // First we need to reconfigure the sink for this sample format.
+            // TODO(jacqueline): We should verify whether or not the sink can
+            // actually deal with this format first.
+            if (current_input_format_ != current_output_format_) {
+              current_output_format_ = current_input_format_;
+              sink_->Configure(*current_output_format_);
+              timer_.reset(new Timer(
+                  std::get<StreamInfo::Pcm>(*current_output_format_)));
+            }
+            // Stream the raw samples directly to the sink.
+            xStreamBufferSend(sink_->stream(), bytes.data(), bytes.size_bytes(),
+                              portMAX_DELAY);
+            timer_->AddBytes(bytes.size_bytes());
+            return bytes.size_bytes();
+          }
+          // Else, assume it's an encoded stream.
+
+          size_t bytes_used = 0;
+          if (is_new_stream_) {
+            // This is a new stream! First order of business is verifying that
+            // we can indeed decode it.
+            auto res = codec_->BeginStream(bytes);
+            bytes_used += res.first;
+
+            if (res.second.has_error()) {
+              if (res.second.error() != codecs::ICodec::Error::kOutOfInput) {
+                // Decoding the header failed, so we can't actually deal with
+                // this stream after all. It could be malformed.
+                ESP_LOGE(kTag, "error beginning stream");
+                codec_.reset();
+              }
+              return bytes_used;
+            }
+            is_new_stream_ = false;
+
+            codecs::ICodec::OutputFormat format = res.second.value();
+            StreamInfo::Pcm pcm{
+                .channels = format.num_channels,
+                .bits_per_sample = format.bits_per_sample,
+                .sample_rate = format.sample_rate_hz,
+            };
+            StreamInfo::Format new_format{pcm};
+            timer_.reset(new Timer{pcm});
+            if (format.duration_seconds) {
+              timer_->SetLengthSeconds(*format.duration_seconds);
+            }
+
+            // Now that we have the output format for decoded samples from this
+            // stream, we need to see if they are compatible with what's already
+            // in the sink stream.
+            if (new_format != current_output_format_) {
+              // The new format is different to the old one. Wait for the sink
+              // to drain before continuing.
+              while (!xStreamBufferIsEmpty(sink_->stream())) {
+                ESP_LOGI(kTag, "waiting for sink stream to drain...");
+                // TODO(jacqueline): Get the sink drain ISR to notify us of this
+                // via semaphore instead of busy-ish waiting.
+                vTaskDelay(pdMS_TO_TICKS(100));
+              }
+            }
+
+            ESP_LOGI(kTag, "configuring sink");
+            current_output_format_ = new_format;
+            sink_->Configure(new_format);
+            timer_.reset(
+                new Timer(std::get<StreamInfo::Pcm>(*current_output_format_)));
+          }
+
+          // At this point the decoder has been initialised, and the sink has
+          // been correctly configured. All that remains is to throw samples
+          // into the sink as fast as possible.
+          while (bytes_used < bytes.size_bytes()) {
+            auto res =
+                codec_->ContinueStream(bytes.subspan(bytes_used),
+                                       {sample_buffer_, sample_buffer_len_});
+
+            bytes_used += res.first;
+
+            if (res.second.has_error()) {
+              return bytes_used;
+            } else {
+              xStreamBufferSend(sink_->stream(), sample_buffer_,
+                                res.second->bytes_written, portMAX_DELAY);
+              timer_->AddBytes(res.second->bytes_written);
+            }
+          }
+
+          return bytes_used;
+        },
+        portMAX_DELAY);
+  }
+}
 
 }  // namespace audio
diff --git a/src/audio/fatfs_audio_input.cpp b/src/audio/fatfs_audio_input.cpp
index da605a40..6a320a5a 100644
--- a/src/audio/fatfs_audio_input.cpp
+++ b/src/audio/fatfs_audio_input.cpp
@@ -5,96 +5,276 @@
  */
 
 #include "fatfs_audio_input.hpp"
+
 #include <stdint.h>
 
 #include <algorithm>
-#include <chrono>
+#include <climits>
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <future>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <variant>
 
-#include "arena.hpp"
-#include "audio_events.hpp"
-#include "audio_fsm.hpp"
 #include "esp_heap_caps.h"
 #include "esp_log.h"
-#include "event_queue.hpp"
 #include "ff.h"
-#include "freertos/portmacro.h"
 
-#include "audio_element.hpp"
-#include "chunk.hpp"
-#include "stream_buffer.hpp"
-#include "stream_event.hpp"
+#include "audio_events.hpp"
+#include "audio_fsm.hpp"
+#include "audio_source.hpp"
+#include "event_queue.hpp"
+#include "freertos/portmacro.h"
+#include "freertos/projdefs.h"
+#include "future_fetcher.hpp"
+#include "span.hpp"
 #include "stream_info.hpp"
-#include "stream_message.hpp"
 #include "tag_parser.hpp"
-#include "track.hpp"
+#include "tasks.hpp"
 #include "types.hpp"
 
 static const char* kTag = "SRC";
 
 namespace audio {
 
-FatfsAudioInput::FatfsAudioInput()
-    : IAudioElement(),
+static constexpr UINT kFileBufferSize = 4096 * 2;
+static constexpr UINT kStreamerBufferSize = 1024;
+
+static StreamBufferHandle_t sForwardDest = nullptr;
+
+auto forward_cb(const BYTE* buf, UINT buf_length) -> UINT {
+  if (buf_length == 0) {
+    return !xStreamBufferIsFull(sForwardDest);
+  } else {
+    return xStreamBufferSend(sForwardDest, buf, buf_length, 0);
+  }
+}
+
+FileStreamer::FileStreamer(StreamBufferHandle_t dest,
+                           SemaphoreHandle_t data_was_read)
+    : control_(xQueueCreate(1, sizeof(Command))),
+      destination_(dest),
+      data_was_read_(data_was_read),
+      has_data_(false),
+      file_(),
+      next_file_() {
+  assert(sForwardDest == nullptr);
+  sForwardDest = dest;
+  tasks::StartPersistent<tasks::Type::kFileStreamer>([this]() { Main(); });
+}
+
+FileStreamer::~FileStreamer() {
+  sForwardDest = nullptr;
+  Command quit = kQuit;
+  xQueueSend(control_, &quit, portMAX_DELAY);
+  vQueueDelete(control_);
+}
+
+auto FileStreamer::Main() -> void {
+  for (;;) {
+    Command cmd;
+    xQueueReceive(control_, &cmd, portMAX_DELAY);
+
+    if (cmd == kQuit) {
+      break;
+    } else if (cmd == kRestart) {
+      CloseFile();
+      xStreamBufferReset(destination_);
+      file_ = std::move(next_file_);
+      has_data_ = file_ != nullptr;
+    } else if (cmd == kRefillBuffer && file_) {
+      UINT bytes_sent = 0;  // Unused.
+      // Use f_forward to push bytes directly from FATFS internal buffers into
+      // the destination. This has the nice side effect of letting FATFS decide
+      // the most efficient way to pull in data from disk; usually one whole
+      // sector at a time. Consult the FATFS lib application notes if changing
+      // this to use f_read.
+      FRESULT res = f_forward(file_.get(), forward_cb, UINT_MAX, &bytes_sent);
+      if (res != FR_OK || f_eof(file_.get())) {
+        CloseFile();
+        has_data_ = false;
+      }
+      if (bytes_sent > 0) {
+        xSemaphoreGive(data_was_read_);
+      }
+    }
+  }
+
+  ESP_LOGW(kTag, "quit file streamer");
+  CloseFile();
+  vTaskDelete(NULL);
+}
+
+auto FileStreamer::Fetch() -> void {
+  if (!has_data_.load()) {
+    return;
+  }
+  Command refill = kRefillBuffer;
+  xQueueSend(control_, &refill, portMAX_DELAY);
+}
+
+auto FileStreamer::HasFinished() -> bool {
+  return !has_data_.load();
+}
+
+auto FileStreamer::Restart(std::unique_ptr<FIL> new_file) -> void {
+  next_file_ = std::move(new_file);
+  Command restart = kRestart;
+  xQueueSend(control_, &restart, portMAX_DELAY);
+  Command fill = kRefillBuffer;
+  xQueueSend(control_, &fill, portMAX_DELAY);
+}
+
+auto FileStreamer::CloseFile() -> void {
+  if (!file_) {
+    return;
+  }
+  ESP_LOGI(kTag, "closing file");
+  f_close(file_.get());
+  file_ = {};
+  events::Dispatch<internal::InputFileClosed, AudioState>({});
+}
+
+FatfsAudioInput::FatfsAudioInput(
+    std::shared_ptr<database::ITagParser> tag_parser)
+    : IAudioSource(),
+      tag_parser_(tag_parser),
+      has_data_(xSemaphoreCreateBinary()),
+      streamer_buffer_(xStreamBufferCreate(kStreamerBufferSize, 1)),
+      streamer_(new FileStreamer(streamer_buffer_, has_data_)),
+      file_buffer_info_(),
+      file_buffer_len_(kFileBufferSize),
+      file_buffer_(reinterpret_cast<std::byte*>(
+          heap_caps_malloc(file_buffer_len_,
+                           MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL))),
+      file_buffer_stream_(&file_buffer_info_, {file_buffer_, file_buffer_len_}),
+      source_mutex_(),
       pending_path_(),
-      current_file_(),
-      is_file_open_(false),
-      has_prepared_output_(false),
-      current_container_(),
       current_format_() {}
 
-FatfsAudioInput::~FatfsAudioInput() {}
+FatfsAudioInput::~FatfsAudioInput() {
+  streamer_.reset();
+  vStreamBufferDelete(streamer_buffer_);
+  vSemaphoreDelete(has_data_);
+  free(file_buffer_);
+}
 
-auto FatfsAudioInput::OpenFile(std::future<std::optional<std::string>>&& path)
+auto FatfsAudioInput::SetPath(std::future<std::optional<std::string>> fut)
     -> void {
-  pending_path_ = std::move(path);
+  std::lock_guard<std::mutex> lock{source_mutex_};
+
+  CloseCurrentFile();
+  pending_path_.reset(
+      new database::FutureFetcher<std::optional<std::string>>(std::move(fut)));
+
+  xSemaphoreGive(has_data_);
 }
 
-auto FatfsAudioInput::OpenFile(const std::string& path) -> bool {
-  current_path_.reset();
-  if (is_file_open_) {
-    f_close(&current_file_);
-    is_file_open_ = false;
-    has_prepared_output_ = false;
-  }
+auto FatfsAudioInput::SetPath(const std::string& path) -> void {
+  std::lock_guard<std::mutex> lock{source_mutex_};
+
+  CloseCurrentFile();
+  OpenFile(path);
+}
+
+auto FatfsAudioInput::SetPath() -> void {
+  std::lock_guard<std::mutex> lock{source_mutex_};
+  CloseCurrentFile();
+}
 
+auto FatfsAudioInput::Read(
+    std::function<bool(StreamInfo::Format)> can_read,
+    std::function<size_t(cpp::span<const std::byte>)> read,
+    TickType_t max_wait) -> void {
+  // Wait until we have data to return.
+  xSemaphoreTake(has_data_, portMAX_DELAY);
+
+  // Ensure the file doesn't change whilst we're trying to get data about it.
+  std::lock_guard<std::mutex> source_lock{source_mutex_};
+
+  // If the path is a future, then wait for it to complete.
+  // TODO(jacqueline): We should really make some kind of FreeRTOS-integrated
+  // way to block a task whilst awaiting a future.
   if (pending_path_) {
-    pending_path_ = {};
+    while (!pending_path_->Finished()) {
+      vTaskDelay(pdMS_TO_TICKS(100));
+    }
+    auto res = pending_path_->Result();
+    pending_path_.reset();
+
+    if (res || *res) {
+      OpenFile(**res);
+    }
+
+    // Bail out now that we've resolved the future. If we end up successfully
+    // readinig from the path, then has_data will be flagged again.
+    return;
+  }
+
+  // Move data from the file streamer's buffer into our file buffer. We need our
+  // own buffer so that we can handle concatenating smaller file chunks into
+  // complete frames for the decoder.
+  OutputStream writer{&file_buffer_stream_};
+  std::size_t bytes_added =
+      xStreamBufferReceive(streamer_buffer_, writer.data().data(),
+                           writer.data().size_bytes(), pdMS_TO_TICKS(0));
+  writer.add(bytes_added);
+
+  // HACK: libmad needs at least MAD_HEADER_GUARD (= 8) extra bytes following a
+  // frame, or else it refuses to decode it.
+  if (IsCurrentFormatMp3() && !HasDataRemaining()) {
+    ESP_LOGI(kTag, "applying MAD_HEADER_GUARD fix");
+    cpp::span<std::byte> buf = writer.data();
+    size_t pad_amount = std::min<size_t>(buf.size_bytes(), 8);
+    std::fill_n(buf.begin(), pad_amount, static_cast<std::byte>(0));
   }
 
+  InputStream reader{&file_buffer_stream_};
+  auto data_for_cb = reader.data();
+  if (!data_for_cb.empty() && std::invoke(can_read, *current_format_)) {
+    reader.consume(std::invoke(read, reader.data()));
+  }
+
+  if (!HasDataRemaining()) {
+    // Out of data. We're finished. Note we don't care about anything left in
+    // the file buffer at this point; the callback as seen it, so if it didn't
+    // consume it then presumably whatever is left isn't enough to form a
+    // complete frame.
+    ESP_LOGI(kTag, "finished streaming file");
+    CloseCurrentFile();
+  } else {
+    // There is still data to be read, or sitting in the buffer.
+    streamer_->Fetch();
+    xSemaphoreGive(has_data_);
+  }
+}
+
+auto FatfsAudioInput::OpenFile(const std::string& path) -> void {
   ESP_LOGI(kTag, "opening file %s", path.c_str());
 
   FILINFO info;
   if (f_stat(path.c_str(), &info) != FR_OK) {
     ESP_LOGE(kTag, "failed to stat file");
+    return;
   }
 
-  database::TagParserImpl tag_parser;
   database::TrackTags tags;
-  if (!tag_parser.ReadAndParseTags(path, &tags)) {
+  if (!tag_parser_->ReadAndParseTags(path, &tags)) {
     ESP_LOGE(kTag, "failed to read tags");
-    return false;
+    return;
   }
 
   auto stream_type = ContainerToStreamType(tags.encoding());
   if (!stream_type.has_value()) {
     ESP_LOGE(kTag, "couldn't match container to stream");
-    return false;
+    return;
   }
 
-  current_container_ = tags.encoding();
-
   if (*stream_type == codecs::StreamType::kPcm && tags.channels &&
       tags.bits_per_sample && tags.channels) {
-    // WAV files are a special case bc they contain raw PCM streams. These don't
-    // need decoding, but we *do* need to parse the PCM format from the header.
-    // TODO(jacqueline): Maybe we should have a decoder for this just to deal
-    // with endianness differences?
     current_format_ = StreamInfo::Pcm{
         .channels = static_cast<uint8_t>(*tags.channels),
         .bits_per_sample = static_cast<uint8_t>(*tags.bits_per_sample),
@@ -107,89 +287,26 @@ auto FatfsAudioInput::OpenFile(const std::string& path) -> bool {
     };
   }
 
-  FRESULT res = f_open(&current_file_, path.c_str(), FA_READ);
+  std::unique_ptr<FIL> file = std::make_unique<FIL>();
+  FRESULT res = f_open(file.get(), path.c_str(), FA_READ);
   if (res != FR_OK) {
     ESP_LOGE(kTag, "failed to open file! res: %i", res);
-    return false;
+    return;
   }
 
+  streamer_->Restart(std::move(file));
+
   events::Dispatch<internal::InputFileOpened, AudioState>({});
-  current_path_ = path;
-  is_file_open_ = true;
-  return true;
 }
 
-auto FatfsAudioInput::NeedsToProcess() const -> bool {
-  return is_file_open_ || pending_path_;
+auto FatfsAudioInput::CloseCurrentFile() -> void {
+  streamer_->Restart({});
+  xStreamBufferReset(streamer_buffer_);
+  current_format_ = {};
 }
 
-auto FatfsAudioInput::Process(const std::vector<InputStream>& inputs,
-                              OutputStream* output) -> void {
-  // If the next path is being given to us asynchronously, then we need to check
-  // in regularly to see if it's available yet.
-  if (pending_path_) {
-    if (!pending_path_->valid()) {
-      pending_path_ = {};
-    } else {
-      if (pending_path_->wait_for(std::chrono::seconds(0)) ==
-          std::future_status::ready) {
-        auto result = pending_path_->get();
-        if (result && result != current_path_) {
-          OpenFile(*result);
-        }
-        pending_path_ = {};
-      }
-    }
-  }
-
-  if (!is_file_open_) {
-    return;
-  }
-
-  // If the output buffer isn't ready for a new stream, then we need to wait.
-  if (!has_prepared_output_ && !output->prepare(*current_format_)) {
-    return;
-  }
-  has_prepared_output_ = true;
-
-  // Performing many small reads is inefficient; it's better to do fewer, larger
-  // reads. Try to achieve this by only reading in new bytes if the output
-  // buffer has been mostly drained.
-  std::size_t max_size = output->data().size_bytes();
-  if (max_size < output->data().size_bytes() / 2) {
-    return;
-  }
-
-  std::size_t size = 0;
-  FRESULT result =
-      f_read(&current_file_, output->data().data(), max_size, &size);
-  if (result != FR_OK) {
-    ESP_LOGE(kTag, "file I/O error %d", result);
-    output->mark_producer_finished();
-    // TODO(jacqueline): Handle errors.
-    return;
-  }
-
-  output->add(size);
-
-  if (size < max_size || f_eof(&current_file_)) {
-    // HACK: In order to decode the last frame of a file, libmad requires 8
-    // 0-bytes ( == MAD_GUARD_BYTES) to be appended to the end of the stream.
-    // It would be better to do this within mad.cpp, but so far it's the only
-    // decoder that has such a requirement.
-    if (current_container_ == database::Encoding::kMp3) {
-      std::fill_n(output->data().begin(), 8, std::byte(0));
-      output->add(8);
-    }
-
-    f_close(&current_file_);
-    is_file_open_ = false;
-    current_path_.reset();
-    has_prepared_output_ = false;
-    output->mark_producer_finished();
-
-    events::Dispatch<internal::InputFileClosed, AudioState>({});
-  }
+auto FatfsAudioInput::HasDataRemaining() -> bool {
+  return !xStreamBufferIsEmpty(streamer_buffer_) || !streamer_->HasFinished();
 }
 
 auto FatfsAudioInput::ContainerToStreamType(database::Encoding enc)
@@ -209,4 +326,15 @@ auto FatfsAudioInput::ContainerToStreamType(database::Encoding enc)
   }
 }
 
+auto FatfsAudioInput::IsCurrentFormatMp3() -> bool {
+  if (!current_format_) {
+    return false;
+  }
+  if (!std::holds_alternative<StreamInfo::Encoded>(*current_format_)) {
+    return false;
+  }
+  return std::get<StreamInfo::Encoded>(*current_format_).type ==
+         codecs::StreamType::kMp3;
+}
+
 }  // namespace audio
diff --git a/src/audio/i2s_audio_output.cpp b/src/audio/i2s_audio_output.cpp
index 5f705dd1..57b5e071 100644
--- a/src/audio/i2s_audio_output.cpp
+++ b/src/audio/i2s_audio_output.cpp
@@ -6,6 +6,7 @@
 
 #include "i2s_audio_output.hpp"
 #include <stdint.h>
+#include <sys/_stdint.h>
 
 #include <algorithm>
 #include <cstddef>
@@ -18,6 +19,7 @@
 #include "audio_element.hpp"
 #include "freertos/projdefs.h"
 #include "gpios.hpp"
+#include "i2c.hpp"
 #include "i2s_dac.hpp"
 #include "result.hpp"
 #include "stream_info.hpp"
@@ -34,7 +36,7 @@ I2SAudioOutput::I2SAudioOutput(drivers::IGpios* expander,
       left_difference_(0),
       attenuation_() {
   SetVolume(25);  // For testing
-  dac_->SetSource(buffer());
+  dac_->SetSource(stream());
 }
 
 I2SAudioOutput::~I2SAudioOutput() {
@@ -68,13 +70,47 @@ auto I2SAudioOutput::GetAdjustedMaxAttenuation() -> int_fast8_t {
   return 0;
 }
 
+static uint8_t vol = 0xFF;
+
 auto I2SAudioOutput::AdjustVolumeUp() -> bool {
-  // TODO
+  vol += 0xF;
+  {
+    drivers::I2CTransaction transaction;
+    transaction.start()
+        .write_addr(0b0011010, I2C_MASTER_WRITE)
+        .write_ack(6, 0b01, vol)
+        .stop();
+    transaction.Execute();
+  }
+  {
+    drivers::I2CTransaction transaction;
+    transaction.start()
+        .write_addr(0b0011010, I2C_MASTER_WRITE)
+        .write_ack(7, 0b11, vol)
+        .stop();
+    transaction.Execute();
+  }
   return true;
 }
 
 auto I2SAudioOutput::AdjustVolumeDown() -> bool {
-  // TODO
+  vol -= 0xF;
+  {
+    drivers::I2CTransaction transaction;
+    transaction.start()
+        .write_addr(0b0011010, I2C_MASTER_WRITE)
+        .write_ack(6, 0b01, vol)
+        .stop();
+    transaction.Execute();
+  }
+  {
+    drivers::I2CTransaction transaction;
+    transaction.start()
+        .write_addr(0b0011010, I2C_MASTER_WRITE)
+        .write_ack(7, 0b11, vol)
+        .stop();
+    transaction.Execute();
+  }
   return true;
 }
 
diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index a6b4754a..e8da415e 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -25,15 +25,12 @@ namespace audio {
  * An audio element that accepts various kinds of encoded audio streams as
  * input, and converts them to uncompressed PCM output.
  */
-class AudioDecoder : public IAudioElement {
+class AudioDecoder {
  public:
   AudioDecoder();
   ~AudioDecoder();
 
-  auto NeedsToProcess() const -> bool override;
-
-  auto Process(const std::vector<InputStream>& inputs, OutputStream* output)
-      -> void override;
+  auto Process(const InputStream& input, OutputStream* output) -> void;
 
   AudioDecoder(const AudioDecoder&) = delete;
   AudioDecoder& operator=(const AudioDecoder&) = delete;
diff --git a/src/audio/include/audio_events.hpp b/src/audio/include/audio_events.hpp
index 8af3703a..933eb7a2 100644
--- a/src/audio/include/audio_events.hpp
+++ b/src/audio/include/audio_events.hpp
@@ -26,7 +26,9 @@ struct PlaybackUpdate : tinyfsm::Event {
   uint32_t seconds_total;
 };
 
-struct QueueUpdate : tinyfsm::Event {};
+struct QueueUpdate : tinyfsm::Event {
+  bool current_changed;
+};
 
 struct VolumeChanged : tinyfsm::Event {};
 
diff --git a/src/audio/include/audio_fsm.hpp b/src/audio/include/audio_fsm.hpp
index 7910f4e2..3a598902 100644
--- a/src/audio/include/audio_fsm.hpp
+++ b/src/audio/include/audio_fsm.hpp
@@ -11,6 +11,7 @@
 #include <vector>
 
 #include "audio_events.hpp"
+#include "audio_task.hpp"
 #include "database.hpp"
 #include "display.hpp"
 #include "fatfs_audio_input.hpp"
@@ -18,6 +19,7 @@
 #include "i2s_audio_output.hpp"
 #include "i2s_dac.hpp"
 #include "storage.hpp"
+#include "tag_parser.hpp"
 #include "tinyfsm.hpp"
 #include "track.hpp"
 
@@ -30,6 +32,7 @@ class AudioState : public tinyfsm::Fsm<AudioState> {
  public:
   static auto Init(drivers::IGpios* gpio_expander,
                    std::weak_ptr<database::Database>,
+                   std::shared_ptr<database::ITagParser>,
                    TrackQueue* queue) -> bool;
 
   virtual ~AudioState() {}
@@ -61,11 +64,12 @@ class AudioState : public tinyfsm::Fsm<AudioState> {
   static std::shared_ptr<drivers::I2SDac> sDac;
   static std::weak_ptr<database::Database> sDatabase;
 
+  static std::unique_ptr<AudioTask> sTask;
   static std::unique_ptr<FatfsAudioInput> sFileSource;
   static std::unique_ptr<I2SAudioOutput> sI2SOutput;
-  static std::vector<std::unique_ptr<IAudioElement>> sPipeline;
 
   static TrackQueue* sTrackQueue;
+  static std::optional<database::TrackId> sCurrentTrack;
 };
 
 namespace states {
diff --git a/src/audio/include/audio_sink.hpp b/src/audio/include/audio_sink.hpp
index ac007bf8..c9124688 100644
--- a/src/audio/include/audio_sink.hpp
+++ b/src/audio/include/audio_sink.hpp
@@ -10,35 +10,25 @@
 #include "audio_element.hpp"
 #include "esp_heap_caps.h"
 #include "freertos/FreeRTOS.h"
+#include "idf_additions.h"
 #include "stream_info.hpp"
+
 namespace audio {
 
 class IAudioSink {
  private:
   // TODO: tune. at least about 12KiB seems right for mp3
-  static const std::size_t kDrainBufferSize = 48 * 1024;
-  uint8_t* buffer_;
-  StaticStreamBuffer_t* metadata_;
-  StreamBufferHandle_t handle_;
+  static const std::size_t kDrainBufferSize = 24 * 1024;
+  StreamBufferHandle_t stream_;
 
  public:
   IAudioSink()
-      : buffer_(reinterpret_cast<uint8_t*>(
-            heap_caps_malloc(kDrainBufferSize,
-                             MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT))),
-        metadata_(reinterpret_cast<StaticStreamBuffer_t*>(
-            heap_caps_malloc(sizeof(StaticStreamBuffer_t),
-                             MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT))),
-        handle_(xStreamBufferCreateStatic(kDrainBufferSize,
-                                          1,
-                                          buffer_,
-                                          metadata_)) {}
-
-  virtual ~IAudioSink() {
-    vStreamBufferDelete(handle_);
-    free(buffer_);
-    free(metadata_);
-  }
+      : stream_(xStreamBufferCreateWithCaps(
+            kDrainBufferSize,
+            1,
+            MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT)) {}
+
+  virtual ~IAudioSink() { vStreamBufferDeleteWithCaps(stream_); }
 
   virtual auto SetInUse(bool) -> void {}
 
@@ -51,7 +41,7 @@ class IAudioSink {
   virtual auto Configure(const StreamInfo::Format& format) -> bool = 0;
   virtual auto Send(const cpp::span<std::byte>& data) -> void = 0;
 
-  auto buffer() -> StreamBufferHandle_t { return handle_; }
+  auto stream() -> StreamBufferHandle_t { return stream_; }
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_source.hpp b/src/audio/include/audio_source.hpp
new file mode 100644
index 00000000..e062fd1a
--- /dev/null
+++ b/src/audio/include/audio_source.hpp
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2023 jacqueline <me@jacqueline.id.au>
+ *
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+
+#pragma once
+
+#include <stdint.h>
+
+#include <memory>
+
+#include "freertos/FreeRTOS.h"
+#include "freertos/portmacro.h"
+#include "freertos/semphr.h"
+
+#include "stream_info.hpp"
+
+namespace audio {
+
+class IAudioSource {
+ public:
+  virtual ~IAudioSource() {}
+
+  /*
+   * Synchronously fetches data from this source.
+   */
+  virtual auto Read(std::function<bool(StreamInfo::Format)>,
+                    std::function<size_t(cpp::span<const std::byte>)>,
+                    TickType_t) -> void = 0;
+};
+
+}  // namespace audio
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index f997caee..e316f17a 100644
--- a/src/audio/include/audio_task.hpp
+++ b/src/audio/include/audio_task.hpp
@@ -6,15 +6,54 @@
 
 #pragma once
 
+#include <sys/_stdint.h>
+#include <cstdint>
+#include <memory>
+#include "audio_decoder.hpp"
 #include "audio_sink.hpp"
+#include "audio_source.hpp"
+#include "codec.hpp"
 #include "pipeline.hpp"
 
 namespace audio {
 
-namespace task {
+class Timer {
+ public:
+  explicit Timer(StreamInfo::Pcm);
 
-auto StartPipeline(Pipeline* pipeline, IAudioSink* sink) -> void;
+  auto SetLengthSeconds(uint32_t) -> void;
+  auto SetLengthBytes(uint32_t) -> void;
 
-}  // namespace task
+  auto AddBytes(std::size_t) -> void;
+
+ private:
+  StreamInfo::Pcm format_;
+
+  uint32_t last_seconds_;
+  uint32_t total_duration_seconds_;
+  float current_seconds_;
+};
+
+class AudioTask {
+ public:
+  static auto Start(IAudioSource* source, IAudioSink* sink) -> AudioTask*;
+
+  auto Main() -> void;
+
+ private:
+  AudioTask(IAudioSource* source, IAudioSink* sink);
+
+  IAudioSource* source_;
+  IAudioSink* sink_;
+  std::unique_ptr<codecs::ICodec> codec_;
+  std::unique_ptr<Timer> timer_;
+
+  bool is_new_stream_;
+  std::optional<StreamInfo::Format> current_input_format_;
+  std::optional<StreamInfo::Format> current_output_format_;
+
+  std::byte* sample_buffer_;
+  std::size_t sample_buffer_len_;
+};
 
 }  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index 56f92fcf..a1b9689b 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -6,57 +6,130 @@
 
 #pragma once
 
+#include <cstddef>
 #include <cstdint>
 #include <future>
 #include <memory>
 #include <string>
-#include <vector>
-
-#include "arena.hpp"
-#include "chunk.hpp"
-#include "freertos/FreeRTOS.h"
 
 #include "ff.h"
-#include "freertos/message_buffer.h"
-#include "freertos/queue.h"
-#include "span.hpp"
-#include "track.hpp"
 
-#include "audio_element.hpp"
-#include "stream_buffer.hpp"
+#include "audio_source.hpp"
+#include "freertos/portmacro.h"
+#include "future_fetcher.hpp"
 #include "stream_info.hpp"
+#include "tag_parser.hpp"
 #include "types.hpp"
 
 namespace audio {
 
-class FatfsAudioInput : public IAudioElement {
+/*
+ * Handles coordination with a persistent background task to asynchronously
+ * read files from disk into a StreamBuffer.
+ */
+class FileStreamer {
  public:
-  FatfsAudioInput();
-  ~FatfsAudioInput();
+  FileStreamer(StreamBufferHandle_t dest, SemaphoreHandle_t first_read);
+  ~FileStreamer();
+
+  /*
+   * Continues reading data into the destination buffer until the destination
+   * is full.
+   */
+  auto Fetch() -> void;
+
+  /* Returns true if the streamer has run out of data from the current file. */
+  auto HasFinished() -> bool;
+
+  /*
+   * Clears any remaining buffered data, and begins reading again from the
+   * given file. This function respects any seeking/reading that has already
+   * been done on the new source file.
+   */
+  auto Restart(std::unique_ptr<FIL>) -> void;
+
+  FileStreamer(const FileStreamer&) = delete;
+  FileStreamer& operator=(const FileStreamer&) = delete;
+
+ private:
+  // Note: private methods here should only be called from the streamer's task.
+
+  auto Main() -> void;
+  auto CloseFile() -> void;
+
+  enum Command {
+    kRestart,
+    kRefillBuffer,
+    kQuit,
+  };
+  QueueHandle_t control_;
+  StreamBufferHandle_t destination_;
+  SemaphoreHandle_t data_was_read_;
+
+  std::atomic<bool> has_data_;
+  std::unique_ptr<FIL> file_;
+  std::unique_ptr<FIL> next_file_;
+};
 
-  auto CurrentFile() -> std::optional<std::string> { return current_path_; }
-  auto OpenFile(std::future<std::optional<std::string>>&& path) -> void;
-  auto OpenFile(const std::string& path) -> bool;
+/*
+ * Audio source that fetches data from a FatFs (or exfat i guess) filesystem.
+ *
+ * All public methods are safe to call from any task.
+ */
+class FatfsAudioInput : public IAudioSource {
+ public:
+  explicit FatfsAudioInput(std::shared_ptr<database::ITagParser> tag_parser);
+  ~FatfsAudioInput();
 
-  auto NeedsToProcess() const -> bool override;
+  /*
+   * Immediately cease reading any current source, and begin reading from the
+   * given file path.
+   */
+  auto SetPath(std::future<std::optional<std::string>>) -> void;
+  auto SetPath(const std::string&) -> void;
+  auto SetPath() -> void;
 
-  auto Process(const std::vector<InputStream>& inputs, OutputStream* output)
-      -> void override;
+  auto Read(std::function<bool(StreamInfo::Format)>,
+            std::function<size_t(cpp::span<const std::byte>)>,
+            TickType_t) -> void override;
 
   FatfsAudioInput(const FatfsAudioInput&) = delete;
   FatfsAudioInput& operator=(const FatfsAudioInput&) = delete;
 
  private:
+  // Note: private methods assume that the appropriate locks have already been
+  // acquired.
+
+  auto OpenFile(const std::string& path) -> void;
+  auto CloseCurrentFile() -> void;
+  auto HasDataRemaining() -> bool;
+
   auto ContainerToStreamType(database::Encoding)
       -> std::optional<codecs::StreamType>;
+  auto IsCurrentFormatMp3() -> bool;
+
+  std::shared_ptr<database::ITagParser> tag_parser_;
+
+  // Semaphore used to block when this source is out of data. This should be
+  // acquired before attempting to read data, and returned after each incomplete
+  // read.
+  SemaphoreHandle_t has_data_;
+
+  StreamBufferHandle_t streamer_buffer_;
+  std::unique_ptr<FileStreamer> streamer_;
+
+  StreamInfo file_buffer_info_;
+  std::size_t file_buffer_len_;
+  std::byte* file_buffer_;
+
+  RawStream file_buffer_stream_;
 
-  std::optional<std::future<std::optional<std::string>>> pending_path_;
-  std::optional<std::string> current_path_;
-  FIL current_file_;
-  bool is_file_open_;
-  bool has_prepared_output_;
+  // Mutex guarding the current file/stream associated with this source. Must be
+  // held during readings, and before altering the current file.
+  std::mutex source_mutex_;
 
-  std::optional<database::Encoding> current_container_;
+  std::unique_ptr<database::FutureFetcher<std::optional<std::string>>>
+      pending_path_;
   std::optional<StreamInfo::Format> current_format_;
 };
 
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index 69bf3c4b..00aa1110 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -15,6 +15,10 @@
 #include <utility>
 #include <variant>
 
+#include "freertos/FreeRTOS.h"
+#include "freertos/ringbuf.h"
+#include "freertos/stream_buffer.h"
+
 #include "result.hpp"
 #include "span.hpp"
 #include "types.hpp"
diff --git a/src/audio/track_queue.cpp b/src/audio/track_queue.cpp
index 0709056f..721329f9 100644
--- a/src/audio/track_queue.cpp
+++ b/src/audio/track_queue.cpp
@@ -81,39 +81,45 @@ auto TrackQueue::GetUpcoming(std::size_t limit) const
 auto TrackQueue::AddNext(database::TrackId t) -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   enqueued_.push_front(t);
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = enqueued_.size() < 2});
 }
 
 auto TrackQueue::AddNext(std::shared_ptr<playlist::ISource> src) -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   enqueued_.push_front(src);
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = enqueued_.size() < 2});
 }
 
 auto TrackQueue::IncludeNext(std::shared_ptr<playlist::IResetableSource> src)
     -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   enqueued_.push_front(src);
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = enqueued_.size() < 2});
 }
 
 auto TrackQueue::AddLast(database::TrackId t) -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   enqueued_.push_back(t);
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = enqueued_.size() < 2});
 }
 
 auto TrackQueue::AddLast(std::shared_ptr<playlist::ISource> src) -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   enqueued_.push_back(src);
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = enqueued_.size() < 2});
 }
 
 auto TrackQueue::IncludeLast(std::shared_ptr<playlist::IResetableSource> src)
     -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   enqueued_.push_back(src);
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = enqueued_.size() < 2});
 }
 
 auto TrackQueue::Next() -> void {
@@ -143,7 +149,8 @@ auto TrackQueue::Next() -> void {
     }
   }
 
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = true});
 }
 
 auto TrackQueue::Previous() -> void {
@@ -173,14 +180,16 @@ auto TrackQueue::Previous() -> void {
   }
   played_.pop_front();
 
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = true});
 }
 
 auto TrackQueue::Clear() -> void {
   const std::lock_guard<std::mutex> lock(mutex_);
   played_.clear();
   enqueued_.clear();
-  events::Dispatch<QueueUpdate, AudioState, ui::UiState>({});
+  events::Dispatch<QueueUpdate, AudioState, ui::UiState>(
+      QueueUpdate{.current_changed = true});
 }
 
 }  // namespace audio
diff --git a/src/codecs/include/codec.hpp b/src/codecs/include/codec.hpp
index 299b16e4..e8be8f0a 100644
--- a/src/codecs/include/codec.hpp
+++ b/src/codecs/include/codec.hpp
@@ -40,6 +40,18 @@ class ICodec {
     kInternalError,
   };
 
+  static auto ErrorString(Error err) -> std::string {
+    switch (err) {
+      case Error::kOutOfInput:
+        return "out of input";
+      case Error::kMalformedData:
+        return "malformed data";
+      case Error::kInternalError:
+        return "internal error";
+    }
+    return "uhh";
+  }
+
   /*
    * Alias for more readable return types. All codec methods, success or
    * failure, should also return the number of bytes they consumed.
diff --git a/src/codecs/mad.cpp b/src/codecs/mad.cpp
index 8b4e2561..29e34a0f 100644
--- a/src/codecs/mad.cpp
+++ b/src/codecs/mad.cpp
@@ -145,11 +145,13 @@ auto MadMp3Decoder::ContinueStream(cpp::span<const std::byte> input,
     for (int channel = 0; channel < synth_.pcm.channels; channel++) {
       uint32_t sample_24 =
           mad_fixed_to_pcm(synth_.pcm.samples[channel][current_sample_], 24);
-      output[output_byte++] = static_cast<std::byte>((sample_24 >> 16) & 0xFF);
-      output[output_byte++] = static_cast<std::byte>((sample_24 >> 8) & 0xFF);
-      output[output_byte++] = static_cast<std::byte>((sample_24)&0xFF);
+
       // 24 bit samples must still be aligned to 32 bits. The LSB is ignored.
       output[output_byte++] = static_cast<std::byte>(0);
+
+      output[output_byte++] = static_cast<std::byte>((sample_24)&0xFF);
+      output[output_byte++] = static_cast<std::byte>((sample_24 >> 8) & 0xFF);
+      output[output_byte++] = static_cast<std::byte>((sample_24 >> 16) & 0xFF);
     }
     current_sample_++;
   }
diff --git a/src/database/include/tag_parser.hpp b/src/database/include/tag_parser.hpp
index b0e9a151..dcc8aa21 100644
--- a/src/database/include/tag_parser.hpp
+++ b/src/database/include/tag_parser.hpp
@@ -26,6 +26,8 @@ class TagParserImpl : public ITagParser {
       -> bool override;
 
  private:
+  std::mutex cache_mutex_;
+
   /*
    * Cache of tags that have already been extracted from files. Ideally this
    * cache should be slightly larger than any page sizes in the UI.
diff --git a/src/database/tag_parser.cpp b/src/database/tag_parser.cpp
index 06d8a8c9..f7e1ceec 100644
--- a/src/database/tag_parser.cpp
+++ b/src/database/tag_parser.cpp
@@ -12,6 +12,7 @@
 #include <tags.h>
 #include <cstdlib>
 #include <iomanip>
+#include <mutex>
 
 namespace database {
 
@@ -97,10 +98,13 @@ static const char* kTag = "TAGS";
 
 auto TagParserImpl::ReadAndParseTags(const std::string& path, TrackTags* out)
     -> bool {
-  std::optional<TrackTags> cached = cache_.Get(path);
-  if (cached) {
-    *out = *cached;
-    return true;
+  {
+    std::lock_guard<std::mutex> lock{cache_mutex_};
+    std::optional<TrackTags> cached = cache_.Get(path);
+    if (cached) {
+      *out = *cached;
+      return true;
+    }
   }
 
   if (path.ends_with(".m4a")) {
@@ -166,7 +170,10 @@ auto TagParserImpl::ReadAndParseTags(const std::string& path, TrackTags* out)
     out->duration = ctx.duration;
   }
 
-  cache_.Put(path, *out);
+  {
+    std::lock_guard<std::mutex> lock{cache_mutex_};
+    cache_.Put(path, *out);
+  }
   return true;
 }
 
diff --git a/src/drivers/i2s_dac.cpp b/src/drivers/i2s_dac.cpp
index 6ffc9e7b..74760543 100644
--- a/src/drivers/i2s_dac.cpp
+++ b/src/drivers/i2s_dac.cpp
@@ -5,7 +5,9 @@
  */
 
 #include "i2s_dac.hpp"
-#include <sys/unistd.h>
+
+#include <stdint.h>
+#include <sys/_stdint.h>
 
 #include <cmath>
 #include <cstdint>
@@ -21,6 +23,7 @@
 #include "esp_log.h"
 #include "freertos/portmacro.h"
 #include "freertos/projdefs.h"
+#include "freertos/ringbuf.h"
 #include "hal/gpio_types.h"
 #include "hal/i2c_types.h"
 
@@ -28,7 +31,6 @@
 #include "hal/i2s_types.h"
 #include "i2c.hpp"
 #include "soc/clk_tree_defs.h"
-#include "sys/_stdint.h"
 
 namespace drivers {
 
@@ -62,14 +64,6 @@ auto I2SDac::create(IGpios* expander) -> std::optional<I2SDac*> {
   i2s_chan_config_t channel_config =
       I2S_CHANNEL_DEFAULT_CONFIG(kI2SPort, I2S_ROLE_MASTER);
 
-  // Use the maximum possible DMA buffer size, since a smaller number of large
-  // copies is faster than a large number of small copies.
-  channel_config.dma_frame_num = 1024;
-  // Triple buffering should be enough to keep samples flowing smoothly.
-  // TODO(jacqueline): verify this with 192kHz 32bps.
-  channel_config.dma_desc_num = 4;
-  // channel_config.auto_clear = true;
-
   ESP_ERROR_CHECK(i2s_new_channel(&channel_config, &i2s_handle, NULL));
   //
   // First, instantiate the instance so it can do all of its power on
@@ -90,7 +84,7 @@ auto I2SDac::create(IGpios* expander) -> std::optional<I2SDac*> {
                        {
                            .mclk_inv = false,
                            .bclk_inv = false,
-                           .ws_inv = true,
+                           .ws_inv = false,
                        }},
   };
 
@@ -107,10 +101,9 @@ I2SDac::I2SDac(IGpios* gpio, i2s_chan_handle_t i2s_handle)
     : gpio_(gpio),
       i2s_handle_(i2s_handle),
       i2s_active_(false),
-      active_page_(),
-      clock_config_(I2S_STD_CLK_DEFAULT_CONFIG(44100)),
+      clock_config_(I2S_STD_CLK_DEFAULT_CONFIG(48000)),
       slot_config_(I2S_STD_PHILIPS_SLOT_DEFAULT_CONFIG(I2S_DATA_BIT_WIDTH_16BIT,
-                                                   I2S_SLOT_MODE_STEREO)) {
+                                                       I2S_SLOT_MODE_STEREO)) {
   clock_config_.clk_src = I2S_CLK_SRC_PLL_160M;
 
   // Keep the 5V circuity off until it's needed.
@@ -122,6 +115,12 @@ I2SDac::I2SDac(IGpios* gpio, i2s_chan_handle_t i2s_handle)
 
   // Power up the charge pump.
   write_register(kPsCtrl, 0, 0b01);
+
+  // TODO: testing
+  // write_register(kDacGainLeft, 0b01, 0x50);
+  // write_register(kDacGainRight, 0b11, 0x50);
+  write_register(kDacGainLeft, 0b01, 0x80);
+  write_register(kDacGainRight, 0b11, 0x78);
 }
 
 I2SDac::~I2SDac() {
@@ -167,14 +166,18 @@ auto I2SDac::Reconfigure(Channels ch, BitsPerSample bps, SampleRate rate)
   switch (bps) {
     case BPS_16:
       slot_config_.data_bit_width = I2S_DATA_BIT_WIDTH_16BIT;
+      slot_config_.ws_width = 16;
       word_length = 0b00;
       break;
     case BPS_24:
       slot_config_.data_bit_width = I2S_DATA_BIT_WIDTH_24BIT;
+      slot_config_.ws_width = 24;
       word_length = 0b10;
       break;
     case BPS_32:
+      // TODO(jacqueline): Error on this? It's not supported anymore.
       slot_config_.data_bit_width = I2S_DATA_BIT_WIDTH_32BIT;
+      slot_config_.ws_width = 32;
       word_length = 0b11;
       break;
   }
@@ -189,9 +192,9 @@ auto I2SDac::Reconfigure(Channels ch, BitsPerSample bps, SampleRate rate)
   ESP_ERROR_CHECK(i2s_channel_reconfig_std_clock(i2s_handle_, &clock_config_));
 
   // Set the correct word size, and set the input format to I2S-justified.
-  write_register(kAifCtrl1, 0, (word_length << 3) & 0b10);
+  write_register(kAifCtrl1, 0, (word_length << 3) | 0b10);
   // Tell the DAC the clock ratio instead of waiting for it to auto detect.
-  write_register(kAifCtrl2, 0, bps == BPS_24 ?  0b100 : 0b011);
+  // write_register(kAifCtrl2, 0, bps == BPS_24 ? 0b100 : 0b011);
 
   if (i2s_active_) {
     i2s_channel_enable(i2s_handle_);
@@ -208,12 +211,6 @@ auto I2SDac::WriteData(const cpp::span<const std::byte>& data) -> void {
   }
 }
 
-static constexpr double increment = (2.0 * 3.141592) / (44100.0 / 500.0);
-static constexpr double amplitude = 16'777'216.0 * 0.6;
-static double current = 0;
-static uint8_t leftover = 0;
-static bool left = false;
-
 extern "C" IRAM_ATTR auto callback(i2s_chan_handle_t handle,
                                    i2s_event_data_t* event,
                                    void* user_ctx) -> bool {
@@ -223,52 +220,20 @@ extern "C" IRAM_ATTR auto callback(i2s_chan_handle_t handle,
   if (event->data == nullptr || event->size == 0) {
     return false;
   }
-  /*
   uint8_t** buf = reinterpret_cast<uint8_t**>(event->data);
-  StreamBufferHandle_t src = reinterpret_cast<StreamBufferHandle_t>(user_ctx);
+  auto src = reinterpret_cast<StreamBufferHandle_t>(user_ctx);
+
   BaseType_t ret = false;
-  std::size_t bytes_received =
+  size_t bytes_written =
       xStreamBufferReceiveFromISR(src, *buf, event->size, &ret);
-  if (bytes_received < event->size) {
-    memset(*buf + bytes_received, 0, event->size - bytes_received);
+
+  // If we ran out of data, then make sure we clear out the DMA buffers rather
+  // than continuing to repreat the last few samples.
+  if (bytes_written < event->size) {
+    std::memset((*buf) + bytes_written, 0, event->size - bytes_written);
   }
+
   return ret;
-  */
-  uint8_t* buf = *(reinterpret_cast<uint8_t**>(event->data));
-  std::size_t i = 0;
-  while (i < event->size) {
-    uint32_t sample = amplitude * std::sin(current);
-    if (leftover > 0) {
-      if (leftover == 2) {
-        buf[i++] = (sample >> 8) & 0xFF;
-        leftover--;
-      }
-      if (leftover == 1) {
-        buf[i++] = sample & 0xFF;
-        leftover--;
-      }
-      continue;
-    }
-
-    buf[i++] = (sample >> 16) & 0xFF;
-    if (i == event->size) {
-      leftover = 2;
-      return false;
-    }
-    buf[i++] = (sample >> 8) & 0xFF;
-    if (i == event->size) {
-      leftover = 1;
-      return false;
-    }
-    buf[i++] = sample & 0xFF;
-    if (left) {
-      current += increment;
-      left = false;
-    } else {
-      left = true;
-    }
-  }
-  return false;
 }
 
 auto I2SDac::SetSource(StreamBufferHandle_t buffer) -> void {
diff --git a/src/drivers/include/i2s_dac.hpp b/src/drivers/include/i2s_dac.hpp
index 39eb9c4c..06c0dc16 100644
--- a/src/drivers/include/i2s_dac.hpp
+++ b/src/drivers/include/i2s_dac.hpp
@@ -18,6 +18,7 @@
 #include "esp_err.h"
 #include "freertos/FreeRTOS.h"
 #include "freertos/portmacro.h"
+#include "freertos/ringbuf.h"
 #include "freertos/stream_buffer.h"
 #include "result.hpp"
 #include "span.hpp"
@@ -73,7 +74,7 @@ class I2SDac {
   IGpios* gpio_;
   i2s_chan_handle_t i2s_handle_;
   bool i2s_active_;
-  std::optional<uint8_t> active_page_;
+  StreamBufferHandle_t buffer_;
 
   i2s_std_clk_config_t clock_config_;
   i2s_std_slot_config_t slot_config_;
diff --git a/src/drivers/include/storage.hpp b/src/drivers/include/storage.hpp
index a9269261..65be75f1 100644
--- a/src/drivers/include/storage.hpp
+++ b/src/drivers/include/storage.hpp
@@ -34,7 +34,6 @@ class SdStorage {
   static auto Create(IGpios* gpio) -> cpp::result<SdStorage*, Error>;
 
   SdStorage(IGpios* gpio,
-            esp_err_t (*do_transaction)(sdspi_dev_handle_t, sdmmc_command_t*),
             sdspi_dev_handle_t handle_,
             std::unique_ptr<sdmmc_host_t> host_,
             std::unique_ptr<sdmmc_card_t> card_,
@@ -47,15 +46,12 @@ class SdStorage {
   auto GetFs() -> FATFS*;
 
   // Not copyable or movable.
-  // TODO: maybe this could be movable?
   SdStorage(const SdStorage&) = delete;
   SdStorage& operator=(const SdStorage&) = delete;
 
  private:
   IGpios* gpio_;
 
-  esp_err_t (*do_transaction_)(sdspi_dev_handle_t, sdmmc_command_t*) = nullptr;
-
   // SPI and SD driver info
   sdspi_dev_handle_t handle_;
   std::unique_ptr<sdmmc_host_t> host_;
diff --git a/src/drivers/spi.cpp b/src/drivers/spi.cpp
index 1e9323ae..b31d1460 100644
--- a/src/drivers/spi.cpp
+++ b/src/drivers/spi.cpp
@@ -38,8 +38,9 @@ esp_err_t init_spi(void) {
       // manages its down use of DMA-capable memory.
       .max_transfer_sz = 128 * 16 * 2,  // TODO: hmm
       .flags = SPICOMMON_BUSFLAG_MASTER | SPICOMMON_BUSFLAG_IOMUX_PINS,
-      .intr_flags =
-          ESP_INTR_FLAG_LOWMED | ESP_INTR_FLAG_SHARED | ESP_INTR_FLAG_IRAM,
+      .intr_flags = ESP_INTR_FLAG_LOWMED | ESP_INTR_FLAG_IRAM,
+      //.intr_flags = ESP_INTR_FLAG_LOWMED | ESP_INTR_FLAG_SHARED |
+      // ESP_INTR_FLAG_IRAM,
   };
 
   if (esp_err_t err = spi_bus_initialize(kSpiHost, &config, SPI_DMA_CH_AUTO)) {
diff --git a/src/drivers/storage.cpp b/src/drivers/storage.cpp
index 04da8819..db257dee 100644
--- a/src/drivers/storage.cpp
+++ b/src/drivers/storage.cpp
@@ -32,37 +32,12 @@ namespace drivers {
 
 const char* kStoragePath = "/sdcard";
 
-// Static functions for interrop with the ESP IDF API, which requires a
-// function pointer.
-namespace callback {
-static std::atomic<SdStorage*> instance = nullptr;
-static std::atomic<esp_err_t (*)(sdspi_dev_handle_t, sdmmc_command_t*)>
-    bootstrap = nullptr;
-
-static esp_err_t do_transaction(sdspi_dev_handle_t handle,
-                                sdmmc_command_t* cmdinfo) {
-  auto bootstrap_fn = bootstrap.load();
-  if (bootstrap_fn != nullptr) {
-    return bootstrap_fn(handle, cmdinfo);
-  }
-  auto instance_unwrapped = instance.load();
-  if (instance_unwrapped == nullptr) {
-    ESP_LOGW(kTag, "uncaught sdspi transaction");
-    return ESP_OK;
-  }
-  // TODO: what if a transaction comes in right now?
-  return instance_unwrapped->HandleTransaction(handle, cmdinfo);
-}
-}  // namespace callback
-
 auto SdStorage::Create(IGpios* gpio) -> cpp::result<SdStorage*, Error> {
   gpio->WriteSync(IGpios::Pin::kSdPowerEnable, 1);
   gpio->WriteSync(IGpios::Pin::kSdMuxSwitch, IGpios::SD_MUX_ESP);
   gpio->WriteSync(IGpios::Pin::kSdMuxDisable, 0);
 
   sdspi_dev_handle_t handle;
-  std::unique_ptr<sdmmc_host_t> host;
-  std::unique_ptr<sdmmc_card_t> card;
   FATFS* fs = nullptr;
 
   // Now we can init the driver and set up the SD card into SPI mode.
@@ -80,17 +55,10 @@ auto SdStorage::Create(IGpios* gpio) -> cpp::result<SdStorage*, Error> {
     return cpp::fail(Error::FAILED_TO_INIT);
   }
 
-  host = std::make_unique<sdmmc_host_t>(sdmmc_host_t SDSPI_HOST_DEFAULT());
-  card = std::make_unique<sdmmc_card_t>();
+  auto host = std::make_unique<sdmmc_host_t>(sdmmc_host_t SDSPI_HOST_DEFAULT());
+  auto card = std::make_unique<sdmmc_card_t>();
 
-  // We manage the CS pin ourselves via the GPIO expander. To do this safely in
-  // a multithreaded environment, we wrap the ESP IDF do_transaction function
-  // with our own that acquires the CS mutex for the duration of the SPI
-  // transaction.
-  auto do_transaction = host->do_transaction;
-  host->do_transaction = &callback::do_transaction;
   host->slot = handle;
-  callback::bootstrap = do_transaction;
 
   // Will return ESP_ERR_INVALID_RESPONSE if there is no card
   esp_err_t err = sdmmc_card_init(host.get(), card.get());
@@ -101,6 +69,7 @@ auto SdStorage::Create(IGpios* gpio) -> cpp::result<SdStorage*, Error> {
 
   ESP_ERROR_CHECK(esp_vfs_fat_register(kStoragePath, "", kMaxOpenFiles, &fs));
   ff_diskio_register_sdmmc(fs->pdrv, card.get());
+  ff_sdmmc_set_disk_status_check(fs->pdrv, true);
 
   // Mount right now, not on first operation.
   FRESULT ferr = f_mount(fs, "", 1);
@@ -109,26 +78,19 @@ auto SdStorage::Create(IGpios* gpio) -> cpp::result<SdStorage*, Error> {
     return cpp::fail(Error::FAILED_TO_MOUNT);
   }
 
-  return new SdStorage(gpio, do_transaction, handle, std::move(host),
-                       std::move(card), fs);
+  return new SdStorage(gpio, handle, std::move(host), std::move(card), fs);
 }
 
 SdStorage::SdStorage(IGpios* gpio,
-                     esp_err_t (*do_transaction)(sdspi_dev_handle_t,
-                                                 sdmmc_command_t*),
                      sdspi_dev_handle_t handle,
                      std::unique_ptr<sdmmc_host_t> host,
                      std::unique_ptr<sdmmc_card_t> card,
                      FATFS* fs)
     : gpio_(gpio),
-      do_transaction_(do_transaction),
       handle_(handle),
       host_(std::move(host)),
       card_(std::move(card)),
-      fs_(fs) {
-  callback::instance = this;
-  callback::bootstrap = nullptr;
-}
+      fs_(fs) {}
 
 SdStorage::~SdStorage() {
   // Unmount and unregister the filesystem
@@ -137,22 +99,14 @@ SdStorage::~SdStorage() {
   esp_vfs_fat_unregister_path(kStoragePath);
   fs_ = nullptr;
 
-  callback::instance = nullptr;
-
   // Uninstall the SPI driver
   sdspi_host_remove_device(this->handle_);
   sdspi_host_deinit();
 
-  gpio_->WriteSync(IGpios::Pin::kSdPowerEnable, 0);
+  gpio_->WriteSync(IGpios::Pin::kSdPowerEnable, 1);
   gpio_->WriteSync(IGpios::Pin::kSdMuxDisable, 1);
 }
 
-auto SdStorage::HandleTransaction(sdspi_dev_handle_t handle,
-                                  sdmmc_command_t* cmdinfo) -> esp_err_t {
-  // TODO: not needed anymore?
-  return do_transaction_(handle, cmdinfo);
-}
-
 auto SdStorage::GetFs() -> FATFS* {
   return fs_;
 }
diff --git a/src/events/event_queue.cpp b/src/events/event_queue.cpp
index 8678211c..8d60218a 100644
--- a/src/events/event_queue.cpp
+++ b/src/events/event_queue.cpp
@@ -16,8 +16,7 @@ static const std::size_t kMaxPendingEvents = 16;
 
 EventQueue::EventQueue()
     : system_handle_(xQueueCreate(kMaxPendingEvents, sizeof(WorkItem*))),
-      ui_handle_(xQueueCreate(kMaxPendingEvents, sizeof(WorkItem*))),
-      audio_handle_(xQueueCreate(kMaxPendingEvents, sizeof(WorkItem*))) {}
+      ui_handle_(xQueueCreate(kMaxPendingEvents, sizeof(WorkItem*))) {}
 
 auto ServiceQueue(QueueHandle_t queue, TickType_t max_wait_time) -> bool {
   WorkItem* item;
@@ -29,7 +28,7 @@ auto ServiceQueue(QueueHandle_t queue, TickType_t max_wait_time) -> bool {
   return false;
 }
 
-auto EventQueue::ServiceSystem(TickType_t max_wait_time) -> bool {
+auto EventQueue::ServiceSystemAndAudio(TickType_t max_wait_time) -> bool {
   return ServiceQueue(system_handle_, max_wait_time);
 }
 
@@ -37,8 +36,4 @@ auto EventQueue::ServiceUi(TickType_t max_wait_time) -> bool {
   return ServiceQueue(ui_handle_, max_wait_time);
 }
 
-auto EventQueue::ServiceAudio(TickType_t max_wait_time) -> bool {
-  return ServiceQueue(audio_handle_, max_wait_time);
-}
-
 }  // namespace events
diff --git a/src/events/include/event_queue.hpp b/src/events/include/event_queue.hpp
index 01f37896..95c331d5 100644
--- a/src/events/include/event_queue.hpp
+++ b/src/events/include/event_queue.hpp
@@ -50,8 +50,6 @@ class EventQueue {
         [=]() { tinyfsm::FsmList<Machine>::template dispatch<Event>(ev); });
     if (std::is_same<Machine, ui::UiState>()) {
       xQueueSend(ui_handle_, &item, portMAX_DELAY);
-    } else if (std::is_same<Machine, audio::AudioState>()) {
-      xQueueSend(audio_handle_, &item, portMAX_DELAY);
     } else {
       xQueueSend(system_handle_, &item, portMAX_DELAY);
     }
@@ -61,9 +59,8 @@ class EventQueue {
   template <typename Event>
   auto Dispatch(const Event& ev) -> void {}
 
-  auto ServiceSystem(TickType_t max_wait_time) -> bool;
+  auto ServiceSystemAndAudio(TickType_t max_wait_time) -> bool;
   auto ServiceUi(TickType_t max_wait_time) -> bool;
-  auto ServiceAudio(TickType_t max_wait_time) -> bool;
 
   EventQueue(EventQueue const&) = delete;
   void operator=(EventQueue const&) = delete;
@@ -73,7 +70,6 @@ class EventQueue {
 
   QueueHandle_t system_handle_;
   QueueHandle_t ui_handle_;
-  QueueHandle_t audio_handle_;
 };
 
 template <typename Event, typename... Machines>
diff --git a/src/main/main.cpp b/src/main/main.cpp
index ac2f6ed8..e2c187b1 100644
--- a/src/main/main.cpp
+++ b/src/main/main.cpp
@@ -19,6 +19,6 @@ extern "C" void app_main(void) {
 
   auto& queue = events::EventQueue::GetInstance();
   while (1) {
-    queue.ServiceSystem(portMAX_DELAY);
+    queue.ServiceSystemAndAudio(portMAX_DELAY);
   }
 }
diff --git a/src/system_fsm/booting.cpp b/src/system_fsm/booting.cpp
index 48b027d2..076f4570 100644
--- a/src/system_fsm/booting.cpp
+++ b/src/system_fsm/booting.cpp
@@ -17,6 +17,7 @@
 #include "spi.hpp"
 #include "system_events.hpp"
 #include "system_fsm.hpp"
+#include "tag_parser.hpp"
 #include "track_queue.hpp"
 #include "ui_fsm.hpp"
 
@@ -50,17 +51,20 @@ auto Booting::entry() -> void {
 
   // Start bringing up LVGL now, since we have all of its prerequisites.
   sTrackQueue.reset(new audio::TrackQueue());
+  /*
   ESP_LOGI(kTag, "starting ui");
   if (!ui::UiState::Init(sGpios.get(), sTrackQueue.get())) {
     events::Dispatch<FatalError, SystemState, ui::UiState, audio::AudioState>(
         FatalError());
     return;
   }
+  */
 
   // Install everything else that is certain to be needed.
   ESP_LOGI(kTag, "installing remaining drivers");
   sSamd.reset(drivers::Samd::Create());
   sBattery.reset(drivers::Battery::Create());
+  sTagParser.reset(new database::TagParserImpl());
 
   if (!sSamd || !sBattery) {
     events::Dispatch<FatalError, SystemState, ui::UiState, audio::AudioState>(
@@ -72,7 +76,8 @@ auto Booting::entry() -> void {
   // state machines and inform them that the system is ready.
 
   ESP_LOGI(kTag, "starting audio");
-  if (!audio::AudioState::Init(sGpios.get(), sDatabase, sTrackQueue.get())) {
+  if (!audio::AudioState::Init(sGpios.get(), sDatabase, sTagParser,
+                               sTrackQueue.get())) {
     events::Dispatch<FatalError, SystemState, ui::UiState, audio::AudioState>(
         FatalError());
     return;
diff --git a/src/system_fsm/include/system_fsm.hpp b/src/system_fsm/include/system_fsm.hpp
index 3c3169d1..03b25156 100644
--- a/src/system_fsm/include/system_fsm.hpp
+++ b/src/system_fsm/include/system_fsm.hpp
@@ -16,6 +16,7 @@
 #include "relative_wheel.hpp"
 #include "samd.hpp"
 #include "storage.hpp"
+#include "tag_parser.hpp"
 #include "tinyfsm.hpp"
 #include "touchwheel.hpp"
 
@@ -57,7 +58,9 @@ class SystemState : public tinyfsm::Fsm<SystemState> {
   static std::shared_ptr<drivers::Battery> sBattery;
   static std::shared_ptr<drivers::SdStorage> sStorage;
   static std::shared_ptr<drivers::Display> sDisplay;
+
   static std::shared_ptr<database::Database> sDatabase;
+  static std::shared_ptr<database::TagParserImpl> sTagParser;
 
   static std::shared_ptr<audio::TrackQueue> sTrackQueue;
 
diff --git a/src/system_fsm/running.cpp b/src/system_fsm/running.cpp
index 1822a071..a46cb8dc 100644
--- a/src/system_fsm/running.cpp
+++ b/src/system_fsm/running.cpp
@@ -5,6 +5,7 @@
  */
 
 #include "app_console.hpp"
+#include "file_gatherer.hpp"
 #include "freertos/projdefs.h"
 #include "result.hpp"
 
@@ -20,6 +21,8 @@ namespace states {
 
 static const char kTag[] = "RUN";
 
+static database::IFileGatherer* sFileGatherer;
+
 /*
  * Ensure the storage and database are both available. If either of these fails
  * to open, then we assume it's an issue with the underlying SD card.
@@ -38,7 +41,8 @@ void Running::entry() {
   vTaskDelay(pdMS_TO_TICKS(250));
 
   ESP_LOGI(kTag, "opening database");
-  auto database_res = database::Database::Open();
+  sFileGatherer = new database::FileGathererImpl();
+  auto database_res = database::Database::Open(sFileGatherer, sTagParser.get());
   if (database_res.has_error()) {
     ESP_LOGW(kTag, "failed to open!");
     events::Dispatch<StorageError, SystemState, audio::AudioState, ui::UiState>(
diff --git a/src/system_fsm/system_fsm.cpp b/src/system_fsm/system_fsm.cpp
index 769d5e4a..c029c6bf 100644
--- a/src/system_fsm/system_fsm.cpp
+++ b/src/system_fsm/system_fsm.cpp
@@ -9,6 +9,7 @@
 #include "event_queue.hpp"
 #include "relative_wheel.hpp"
 #include "system_events.hpp"
+#include "tag_parser.hpp"
 #include "track_queue.hpp"
 
 namespace system_fsm {
@@ -21,7 +22,9 @@ std::shared_ptr<drivers::RelativeWheel> SystemState::sRelativeTouch;
 std::shared_ptr<drivers::Battery> SystemState::sBattery;
 std::shared_ptr<drivers::SdStorage> SystemState::sStorage;
 std::shared_ptr<drivers::Display> SystemState::sDisplay;
+
 std::shared_ptr<database::Database> SystemState::sDatabase;
+std::shared_ptr<database::TagParserImpl> SystemState::sTagParser;
 
 std::shared_ptr<audio::TrackQueue> SystemState::sTrackQueue;
 
@@ -37,14 +40,14 @@ void SystemState::react(const internal::GpioInterrupt& ev) {
   bool prev_key_up = sGpios->Get(drivers::Gpios::Pin::kKeyUp);
   bool prev_key_down = sGpios->Get(drivers::Gpios::Pin::kKeyDown);
   bool prev_key_lock = sGpios->Get(drivers::Gpios::Pin::kKeyLock);
-  bool prev_has_headphones = sGpios->Get(drivers::Gpios::Pin::kPhoneDetect);
+  bool prev_has_headphones = !sGpios->Get(drivers::Gpios::Pin::kPhoneDetect);
 
   sGpios->Read();
 
   bool key_up = sGpios->Get(drivers::Gpios::Pin::kKeyUp);
   bool key_down = sGpios->Get(drivers::Gpios::Pin::kKeyDown);
   bool key_lock = sGpios->Get(drivers::Gpios::Pin::kKeyLock);
-  bool has_headphones = sGpios->Get(drivers::Gpios::Pin::kPhoneDetect);
+  bool has_headphones = !sGpios->Get(drivers::Gpios::Pin::kPhoneDetect);
 
   if (key_up != prev_key_up) {
     events::Dispatch<KeyUpChanged, audio::AudioState, ui::UiState>(
diff --git a/src/tasks/tasks.cpp b/src/tasks/tasks.cpp
index 7365813e..abce0bde 100644
--- a/src/tasks/tasks.cpp
+++ b/src/tasks/tasks.cpp
@@ -26,6 +26,10 @@ auto Name<Type::kUiFlush>() -> std::string {
   return "DISPLAY";
 }
 template <>
+auto Name<Type::kFileStreamer>() -> std::string {
+  return "FSTREAMER";
+}
+template <>
 auto Name<Type::kAudio>() -> std::string {
   return "AUDIO";
 }
@@ -65,6 +69,14 @@ auto AllocateStack<Type::kUiFlush>() -> cpp::span<StackType_t> {
   return {static_cast<StackType_t*>(heap_caps_malloc(size, MALLOC_CAP_DEFAULT)),
           size};
 }
+
+template <>
+auto AllocateStack<Type::kFileStreamer>() -> cpp::span<StackType_t> {
+  std::size_t size = 4 * 1024;
+  return {static_cast<StackType_t*>(heap_caps_malloc(size, MALLOC_CAP_SPIRAM)),
+          size};
+}
+
 // Leveldb is designed for non-embedded use cases, where stack space isn't so
 // much of a concern. It therefore uses an eye-wateringly large amount of stack.
 template <>
@@ -110,6 +122,11 @@ template <>
 auto Priority<Type::kUiFlush>() -> UBaseType_t {
   return 9;
 }
+
+template <>
+auto Priority<Type::kFileStreamer>() -> UBaseType_t {
+  return 10;
+}
 // Database interactions are all inherently async already, due to their
 // potential for disk access. The user likely won't notice or care about a
 // couple of ms extra delay due to scheduling, so give this task the lowest
diff --git a/src/tasks/tasks.hpp b/src/tasks/tasks.hpp
index 4e5dfd17..742bb3cc 100644
--- a/src/tasks/tasks.hpp
+++ b/src/tasks/tasks.hpp
@@ -32,6 +32,8 @@ enum class Type {
   kUi,
   // Task for flushing graphics buffers to the display.
   kUiFlush,
+  // TODO.
+  kFileStreamer,
   // The main audio pipeline task.
   kAudio,
   // Task for running database queries.
@@ -55,9 +57,9 @@ template <Type t>
 auto StartPersistent(const std::function<void(void)>& fn) -> void {
   StaticTask_t* task_buffer = new StaticTask_t;
   cpp::span<StackType_t> stack = AllocateStack<t>();
-  xTaskCreateStatic(&PersistentMain, Name<t>().c_str(), stack.size(),
-                    new std::function<void(void)>(fn), Priority<t>(),
-                    stack.data(), task_buffer);
+  xTaskCreateStaticPinnedToCore(&PersistentMain, Name<t>().c_str(),
+                                stack.size(), new std::function<void(void)>(fn),
+                                Priority<t>(), stack.data(), task_buffer, 0);
 }
 
 class Worker {
diff --git a/tools/cmake/common.cmake b/tools/cmake/common.cmake
index a94f498f..ec403e87 100644
--- a/tools/cmake/common.cmake
+++ b/tools/cmake/common.cmake
@@ -11,15 +11,16 @@ set(COMPONENTS "")
 # External dependencies
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/catch2")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/cbor")
+list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/fatfs")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/komihash")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/libfoxenflac")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/libmad")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/libtags")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/lvgl")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/result")
+list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/shared_string")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/span")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/stb_vorbis")
-list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/shared_string")
 list(APPEND EXTRA_COMPONENT_DIRS "$ENV{PROJ_PATH}/lib/tinyfsm")
 
 include($ENV{IDF_PATH}/tools/cmake/project.cmake)