Rakka/Storage.hs

   1 module Rakka.Storage
   2     ( Storage
   3
   4     , mkStorage -- private
   5
   6     , getPage
   7     , putPage
   8
   9     , getPageA
  10     , putPageA
  11
  12     , searchPages
  13     )
  14     where
  15
  16 import           Control.Arrow.ArrowIO
  17 import           Control.Concurrent
  18 import           Control.Concurrent.STM
  19 import           Control.Exception
  20 import           Control.Monad
  21 import           Control.Monad.Trans
  22 import           Data.Maybe
  23 import           Data.Set (Set)
  24 import qualified Data.Set as S
  25 import           Network.URI
  26 import           Rakka.Page
  27 import           Rakka.Storage.DefaultPage
  28 import           Subversion.Types
  29 import           System.Directory
  30 import           System.FilePath
  31 import           System.IO
  32 import           System.Log.Logger
  33 import           System.Posix.Files
  34 import           System.Posix.Types
  35 import           System.Posix.IO
  36 import           Subversion.FileSystem
  37 import           Subversion.Repository
  38 import           Text.HyperEstraier hiding (WriteLock)
  39
  40 logger = "Rakka.Storage"
  41
  42
  43 data Storage
  44     = Storage {
  45         stoRepository :: !Repository
  46       , stoIndexChan  :: !(TChan IndexReq)
  47       }
  48
  49
  50 data IndexReq
  51     = SyncIndex
  52     | SearchIndex !Condition !(TMVar [(PageName, RevNum)])
  53
  54
  55 mkStorage :: FilePath -> Repository -> (Page -> IO Document) -> IO Storage
  56 mkStorage lsdir repos mkDraft
  57     = do chan <- startIndexManager lsdir repos mkDraft
  58          let sto = Storage {
  59                      stoRepository = repos
  60                    , stoIndexChan  = chan
  61                    }
  62          syncIndex sto
  63          return sto
  64
  65
  66 getPage :: MonadIO m => Storage -> PageName -> Maybe RevNum -> m (Maybe Page)
  67 getPage = ((liftIO .) .) . getPage' . stoRepository
  68
  69
  70 getPage' :: Repository -> PageName -> Maybe RevNum -> IO (Maybe Page)
  71 getPage' repos name rev
  72     = loadDefaultPage name -- FIXME
  73
  74
  75 putPage :: MonadIO m => Storage -> Page -> RevNum -> m ()
  76 putPage sto page oldRev
  77     = error "FIXME: not implemented"
  78
  79
  80 getPageA :: ArrowIO a => Storage -> a (PageName, Maybe RevNum) (Maybe Page)
  81 getPageA = arrIO2 . getPage
  82
  83
  84 putPageA :: ArrowIO a => Storage -> a (Page, RevNum) ()
  85 putPageA = arrIO2 . putPage
  86
  87
  88 searchPages :: MonadIO m => Storage -> Condition -> m [(PageName, RevNum)]
  89 searchPages sto cond
  90     = liftIO $
  91       do var <- newEmptyTMVarIO
  92          atomically $ writeTChan (stoIndexChan sto) (SearchIndex cond var)
  93          atomically $ takeTMVar var
  94
  95
  96 syncIndex :: Storage -> IO ()
  97 syncIndex sto
  98     = atomically $ writeTChan (stoIndexChan sto) SyncIndex
  99
 100
 101 findAllPages :: Repository -> RevNum -> IO (Set PageName)
 102 findAllPages _     0   = findAllDefaultPages
 103 findAllPages repos rev
 104     = findAllDefaultPages -- FIXME
 105
 106
 107 findChangedPages :: Repository -> RevNum -> RevNum -> IO (Set PageName)
 108 findChangedPages repos 0      newRev = findAllPages repos newRev
 109 findChangedPages repos oldRev newRev
 110     = fail "FIXME: not impl"
 111
 112
 113 getCurrentRevNum :: Repository -> IO RevNum
 114 getCurrentRevNum repos
 115     = getRepositoryFS repos >>= getYoungestRev
 116
 117
 118 startIndexManager :: FilePath -> Repository -> (Page -> IO Document) -> IO (TChan IndexReq)
 119 startIndexManager lsdir repos mkDraft
 120     = do chan  <- newTChanIO
 121          index <- openIndex indexDir revFile
 122          forkIO (loop chan index)
 123          return chan
 124     where
 125       indexDir = lsdir </> "index"
 126       revFile  = lsdir </> "indexRev"
 127
 128       loop :: TChan IndexReq -> Database -> IO ()
 129       loop chan index
 130           = do req <- atomically $ readTChan chan
 131                case req of
 132                  SyncIndex
 133                      -> syncIndex' index revFile repos mkDraft
 134                  SearchIndex cond var
 135                      -> do result <- searchIndex index cond
 136                            atomically $ putTMVar var result
 137                loop chan index
 138
 139
 140 -- casket を R/W モードで開く。成功したらそのまま返し、失敗したら
 141 -- indexDir と revFile を削除してから casket を R/W モードで開く。
 142 openIndex :: FilePath -> FilePath -> IO Database
 143 openIndex indexDir revFile
 144     = do ret <- openDatabase indexDir (Writer [])
 145          case ret of
 146            Right index
 147                -> do debugM logger ("Opened an H.E. index on " ++ indexDir)
 148                      return index
 149
 150            Left err
 151                -> do warningM logger ("Failed to open an H.E. index on "
 152                                       ++ indexDir ++ ": " ++ show err)
 153
 154                      indexExists <- doesDirectoryExist indexDir
 155                      when indexExists
 156                               $ removeDirectoryRecursive indexDir
 157
 158                      revFileExists <- doesFileExist revFile
 159                      when revFileExists
 160                               $ removeFile revFile
 161
 162                      Right index <- openDatabase indexDir (Writer [Create []])
 163                      addAttrIndex index "@uri"           SeqIndex
 164                      addAttrIndex index "rakka:revision" SeqIndex
 165                      noticeM logger ("Created an H.E. index on " ++ indexDir)
 166
 167                      return index
 168
 169
 170 syncIndex' :: Database -> FilePath -> Repository -> (Page -> IO Document) -> IO ()
 171 syncIndex' index revFile repos mkDraft
 172     = updateIndexRev revFile $ \ oldRev ->
 173       do debugM logger ("The index revision is currently " ++ show oldRev)
 174
 175          newRev <- getCurrentRevNum repos
 176          debugM logger ("The repository revision is currently " ++ show newRev)
 177
 178          when (newRev /= oldRev) (syncIndex'' oldRev newRev)
 179          return newRev
 180     where
 181       syncIndex'' :: RevNum -> RevNum -> IO ()
 182       syncIndex'' oldRev newRev
 183           = do pages <- findChangedPages repos oldRev newRev
 184                mapM_ (updateIndex index repos mkDraft newRev) (S.toList pages)
 185
 186
 187 searchIndex :: Database -> Condition -> IO [(PageName, RevNum)]
 188 searchIndex index cond
 189     = searchDatabase index cond >>= mapM fromId
 190     where
 191       fromId :: DocumentID -> IO (PageName, RevNum)
 192       fromId docId
 193           = do uri <- getDocURI index docId
 194                rev <- getDocAttr index docId "rakka:revision"
 195                       >>= return . read . fromJust
 196                return (decodePageName $ uriPath uri, rev)
 197
 198
 199 updateIndex :: Database
 200             -> Repository
 201             -> (Page -> IO Document)
 202             -> RevNum
 203             -> PageName
 204             -> IO ()
 205 updateIndex index repos mkDraft rev name
 206     = do pageM <- getPage' repos name (Just rev)
 207          case pageM of
 208            -- ページが削除された
 209            Nothing
 210                -> do docIdM <- getDocIdByURI index (mkRakkaURI name)
 211                      case docIdM of
 212                        Nothing    -> return ()
 213                        Just docId -> do removeDocument index docId [CleaningRemove]
 214                                         infoM logger ("Removed page " ++ name ++ " from the index")
 215            Just page
 216                -> do draft <- mkDraft page
 217                      putDocument index draft [CleaningPut]
 218                      infoM logger ("Indexed page " ++ name ++ " of revision " ++ show (pageRevision page))
 219
 220
 221 updateIndexRev :: FilePath -> (RevNum -> IO RevNum) -> IO ()
 222 updateIndexRev revFile f = bracket acquireLock releaseLock update
 223     where
 224       acquireLock :: IO Fd
 225       acquireLock
 226           = do fd <- openFd revFile ReadWrite (Just stdFileMode) defaultFileFlags
 227                waitToSetLock fd (WriteLock, AbsoluteSeek, 0, 0)
 228                return fd
 229
 230       releaseLock :: Fd -> IO ()
 231       releaseLock fd
 232           = setLock fd (Unlock, AbsoluteSeek, 0, 0)
 233
 234       update :: Fd -> IO ()
 235       update fd
 236           = do fdSeek fd AbsoluteSeek 0
 237                size <- return . fromIntegral . fileSize =<< getFdStatus fd
 238                (revStr, gotSize) <- fdRead fd size
 239                when (size /= gotSize) $ fail ("read " ++ show gotSize ++
 240                                               " bytes but expected " ++ show size ++ " bytes")
 241
 242                let rev = case revStr of
 243                            "" -> 0
 244                            _  -> read revStr
 245
 246                rev' <- f rev
 247
 248                let revStr' = show rev' ++ "\n"
 249                    size'   = fromIntegral $ length revStr'
 250
 251                fdSeek fd AbsoluteSeek 0
 252                setFdSize fd 0
 253                wroteSize <- fdWrite fd revStr'
 254                when (size' /= wroteSize) $ fail ("wrote " ++ show wroteSize ++
 255                                                  " bytes but expected " ++ show size' ++ " bytes")